diff --git a/python/x402/http/facilitator_client.py b/python/x402/http/facilitator_client.py
index 3728ceb01a..adfd60ab10 100644
--- a/python/x402/http/facilitator_client.py
+++ b/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,
diff --git a/python/x402/http/facilitator_client_base.py b/python/x402/http/facilitator_client_base.py
index 51ed934513..8b54828ce0 100644
--- a/python/x402/http/facilitator_client_base.py
+++ b/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:
diff --git a/python/x402/http/middleware/flask.py b/python/x402/http/middleware/flask.py
index da3e3b2b29..fc34d47dd3 100644
--- a/python/x402/http/middleware/flask.py
+++ b/python/x402/http/middleware/flask.py
@@ -21,6 +21,7 @@
 import logging
 import re
 import threading
+import time
 from collections.abc import Callable, Iterator
 from typing import TYPE_CHECKING, Any
 
@@ -42,7 +43,7 @@
 from ..x402_http_server import PaywallProvider, x402HTTPResourceServerSync
 from ...schemas import PaymentPayload, PaymentRequirements, SettlementOverrides
 from ...schemas.v1 import PaymentPayloadV1
-from ...session import SessionStoreProtocol, UptoSession
+from ...session import SessionStoreProtocol, UptoSession, signed_authorization_deadline
 
 if TYPE_CHECKING:
     from ...server import x402ResourceServerSync
@@ -769,6 +770,7 @@ def __init__(
         session_cost_calculator: Callable[[dict[str, Any]], int] | None = None,
         cost_per_request: int | None = None,
         session_idle_timeout: int = 3600,
+        settlement_safety_margin: int = 60,
     ) -> None:
         """Initialize Flask payment middleware.
 
@@ -784,6 +786,13 @@ def __init__(
                 computing per-request cost in token smallest units.
             cost_per_request: Static fallback cost when calculator is absent.
             session_idle_timeout: Seconds before an idle session is settled.
+            settlement_safety_margin: Seconds before a session's signed
+                authorization deadline at which the reaper force-settles it
+                (and stops accepting new draw-downs against it). Guards against
+                a long-lived session outliving its own Permit2/EIP-3009
+                deadline, which would make the on-chain settlement revert and
+                silently drop the tab. Must be smaller than the advertised
+                ``max_timeout_seconds`` for a route.
         """
         # Auto-register bazaar extension if routes declare it
         if _check_if_bazaar_needed(routes):
@@ -805,6 +814,7 @@ def __init__(
         self._session_cost_calculator = session_cost_calculator
         self._cost_per_request = cost_per_request
         self._session_idle_timeout = session_idle_timeout
+        self._settlement_safety_margin = max(0, int(settlement_safety_margin))
         self._payment_to_session: dict[str, str] = {}
         self._session_map_lock = threading.Lock()
         self._reaper_thread: threading.Thread | None = None
@@ -960,8 +970,18 @@ def _handle_session_mode(
 
         if session_id:
             session = store.get_session(session_id)
-            if session is None or session.settled or session.is_exhausted:
-                # Session gone or used up — require new payment
+            if (
+                session is None
+                or session.settled
+                or session.settling
+                or session.is_exhausted
+                or self._session_needs_resign(session)
+            ):
+                # Session gone, being settled, used up, or nearing its
+                # authorization deadline — require a new payment so the client
+                # re-signs a fresh window. (settling must be terminal for reuse:
+                # add_cost() rejects settling sessions, so reusing one would
+                # serve a response without charging.)
                 start_response(
                     "402 Payment Required",
                     [("Content-Type", "application/json")],
@@ -972,6 +992,18 @@ def _handle_session_mode(
             payload_dict = result.payment_payload.model_dump(by_alias=True)
             reqs_dict = result.payment_requirements.model_dump(by_alias=True)
             max_amount = int(result.payment_requirements.amount)
+            if not self._authorization_admissible(payload_dict):
+                # Refuse to open a draw-down session we cannot safely settle.
+                # We do NOT stream (and thus do not perform paid work) here.
+                start_response(
+                    "402 Payment Required",
+                    [("Content-Type", "application/json")],
+                )
+                return [
+                    json.dumps(
+                        {"error": "Payment authorization deadline missing or too soon"}
+                    ).encode()
+                ]
             session_id = store.create_session(
                 permit_payload=payload_dict,
                 requirements=reqs_dict,
@@ -999,6 +1031,48 @@ def _handle_session_mode(
         )
 
 
+    def _authorization_admissible(self, permit_payload: dict[str, Any]) -> bool:
+        """Whether a fresh authorization may open a draw-down session.
+
+        Strict / fail-closed — this gates paid work, so anything we cannot prove
+        safe is rejected:
+
+        * **Signed deadline required.** We admit only if the payload carries a
+          parseable *signed* deadline (Permit2 ``deadline`` / EIP-3009
+          ``validBefore``). We never fall back to ``created_at +
+          maxTimeoutSeconds`` for admission: that guess can outlive the true
+          on-chain deadline, so settlement would revert and the tab would be
+          lost. A malformed/absent deadline is a rejection, not an assumption.
+        * **Enough runway.** The signed deadline must be further out than the
+          settlement safety margin. Otherwise the session would be force-settled
+          (and require a re-sign) almost immediately, letting a client drive a
+          rapid settle/re-sign churn — each settlement a full-gas on-chain tx —
+          with an already-near-expiry authorization.
+        """
+        deadline = signed_authorization_deadline(permit_payload)
+        if deadline is None:
+            return False
+        return time.time() < deadline - self._settlement_safety_margin
+
+    def _session_needs_resign(self, session: UptoSession) -> bool:
+        """Whether a session is too close to its authorization deadline to reuse.
+
+        Once within the settlement safety margin, the reaper will force-settle
+        the accumulated tab; we must stop accepting new draw-downs so the
+        settled amount is final and the client re-signs a fresh authorization
+        (fresh deadline) instead of piling cost onto an expiring one.
+
+        Keys off the *signed* deadline only (fail closed): reuse serves new paid
+        work, so we refuse rather than keep drawing down an authorization whose
+        on-chain deadline we cannot verify. The reaper still uses
+        ``settlement_deadline`` (with its advertised-window fallback) to settle
+        whatever tab already accumulated.
+        """
+        deadline = session.signed_deadline
+        if deadline is None:
+            return True
+        return time.time() >= deadline - self._settlement_safety_margin
+
     def _resume_session_request(
         self,
         environ: dict[str, Any],
@@ -1019,6 +1093,7 @@ def _resume_session_request(
             or session.settled
             or session.settling
             or session.is_exhausted
+            or self._session_needs_resign(session)
             or (session.route_method and session.route_method != context.method)
             or (session.route_path and session.route_path != context.path)
         ):
@@ -1481,7 +1556,10 @@ def _start_reaper(self) -> None:
 
     def _reaper_loop(self) -> None:
         """Background loop that periodically settles ready sessions."""
-        interval = min(self._session_idle_timeout // 4, 30)
+        # Cadence must be fine-grained enough to catch a session before its
+        # authorization deadline, so it also tracks the safety margin.
+        cadence_basis = min(self._session_idle_timeout, self._settlement_safety_margin)
+        interval = min(cadence_basis // 4, 30)
         interval = max(interval, 5)
         while not self._reaper_stop.wait(timeout=interval):
             try:
@@ -1490,11 +1568,17 @@ def _reaper_loop(self) -> None:
                 logger.exception("Error in session reaper cycle")
 
     def _settle_ready_sessions(self) -> None:
-        """Settle all expired and exhausted sessions."""
+        """Settle all deadline-due, expired, and exhausted sessions."""
         store = self._session_store
         if store is None:
             return
 
+        # Deadline-approaching sessions first: they MUST settle before their
+        # signed authorization expires or the on-chain settle reverts.
+        get_due = getattr(store, "get_settlement_due_sessions", None)
+        if callable(get_due):
+            for session in get_due(self._settlement_safety_margin):
+                self._settle_session(session)
         for session in store.get_expired_sessions(self._session_idle_timeout):
             self._settle_session(session)
         for session in store.get_exhausted_sessions():
@@ -1589,6 +1673,7 @@ def payment_middleware(
     session_cost_calculator: Callable[[dict[str, Any]], int] | None = None,
     cost_per_request: int | None = None,
     session_idle_timeout: int = 3600,
+    settlement_safety_margin: int = 60,
 ) -> PaymentMiddleware:
     """Create Flask payment middleware with pre-configured server.
 
@@ -1603,6 +1688,8 @@ def payment_middleware(
         session_cost_calculator: Callback for per-request cost calculation.
         cost_per_request: Static fallback cost per request.
         session_idle_timeout: Idle timeout before session is settled.
+        settlement_safety_margin: Seconds before the signed authorization
+            deadline at which the reaper force-settles a session.
 
     Returns:
         PaymentMiddleware instance.
@@ -1618,6 +1705,7 @@ def payment_middleware(
         session_cost_calculator=session_cost_calculator,
         cost_per_request=cost_per_request,
         session_idle_timeout=session_idle_timeout,
+        settlement_safety_margin=settlement_safety_margin,
     )
 
 
diff --git a/python/x402/redis_session.py b/python/x402/redis_session.py
index 5815cf1d5a..021b01cfd6 100644
--- a/python/x402/redis_session.py
+++ b/python/x402/redis_session.py
@@ -218,6 +218,25 @@ def get_exhausted_sessions(self) -> list[UptoSession]:
                     exhausted.append(session)
         return exhausted
 
+    def get_settlement_due_sessions(
+        self, safety_margin_seconds: int
+    ) -> list[UptoSession]:
+        now = time.time()
+        due: list[UptoSession] = []
+        for session in self._iter_active_sessions():
+            if session.settled:
+                continue
+            deadline = session.settlement_deadline
+            if deadline is not None and now >= deadline - safety_margin_seconds:
+                # Atomically claim to prevent duplicate settlement
+                claimed = self._claim_script(
+                    keys=[self._session_key(session.session_id)],
+                    args=[],
+                )
+                if claimed is not None:
+                    due.append(session)
+        return due
+
     @property
     def active_count(self) -> int:
         count = 0
diff --git a/python/x402/session.py b/python/x402/session.py
index cadcf627e4..c149ffbee4 100644
--- a/python/x402/session.py
+++ b/python/x402/session.py
@@ -13,6 +13,37 @@
 from typing import Any, Protocol, runtime_checkable
 
 
+def signed_authorization_deadline(permit_payload: dict[str, Any]) -> float | None:
+    """Extract the signed on-chain deadline from an upto payment payload.
+
+    Reads the deadline embedded in (and covered by) the client's signature —
+    Permit2 ``deadline`` or EIP-3009 ``validBefore`` — at its canonical location
+    under ``payload.permit2Authorization`` / ``payload.authorization``.
+
+    This is the *authoritative* deadline: settling after it reverts on-chain.
+    It is deliberately kept separate from any advertised-window fallback so
+    callers that must fail closed (e.g. refusing to open a session we cannot
+    prove a settlement deadline for) can tell "signed value present" apart from
+    "guessed from requirements".
+
+    Returns:
+        Unix timestamp (seconds) of the signed deadline, or ``None`` if it is
+        absent or unparseable.
+    """
+    try:
+        inner = permit_payload.get("payload", {})
+        auth = inner.get("permit2Authorization") or inner.get("authorization")
+        if isinstance(auth, dict):
+            raw = auth.get("deadline")
+            if raw is None:
+                raw = auth.get("validBefore")
+            if raw is not None:
+                return float(int(raw))
+    except (AttributeError, TypeError, ValueError):
+        pass
+    return None
+
+
 @dataclass
 class UptoSession:
     """An active upto payment session.
@@ -52,6 +83,57 @@ def is_exhausted(self) -> bool:
         """Whether the spend cap has been fully consumed."""
         return self.accumulated_cost >= self.max_amount
 
+    @property
+    def signed_deadline(self) -> float | None:
+        """The deadline taken strictly from the signed payment payload.
+
+        ``None`` when the payload does not expose a parseable Permit2
+        ``deadline`` / EIP-3009 ``validBefore``. Callers that must not settle
+        past an authorization they cannot verify should refuse a session whose
+        ``signed_deadline`` is ``None`` rather than trust ``settlement_deadline``
+        (which may substitute an advertised-window guess).
+        """
+        return signed_authorization_deadline(self.permit_payload)
+
+    @property
+    def settlement_deadline(self) -> float | None:
+        """Absolute unix time by which this session MUST be settled on-chain.
+
+        A draw-down session reuses a single signed payment authorization whose
+        deadline is fixed at signing time and never refreshed. If we settle
+        after that deadline the on-chain ``settle`` reverts (e.g. Permit2
+        ``deadline`` expired / EIP-3009 ``validBefore`` passed) and the whole
+        accumulated tab is silently lost. This returns the hard deadline so the
+        reaper can force settlement *before* it, preferring the exact value
+        embedded in the signed payload and falling back to
+        ``created_at + max validity window`` from the requirements.
+
+        Note: the fallback is a best-effort *guess* and can be later than the
+        true signed deadline; do not rely on it for admission decisions — use
+        ``signed_deadline`` for those. See ``PaymentMiddleware`` session
+        creation.
+
+        Returns:
+            Unix timestamp (seconds) of the authorization deadline, or ``None``
+            if it cannot be determined.
+        """
+        # Prefer the exact signed deadline from the payment payload.
+        signed = self.signed_deadline
+        if signed is not None:
+            return signed
+
+        # Fall back to created_at + the requirement's advertised validity window.
+        max_timeout = self.requirements.get("maxTimeoutSeconds")
+        if max_timeout is None:
+            max_timeout = self.requirements.get("max_timeout_seconds")
+        try:
+            if max_timeout is not None:
+                return self.created_at + float(int(max_timeout))
+        except (TypeError, ValueError):
+            pass
+
+        return None
+
     def to_dict(self) -> dict[str, Any]:
         """Serialize session to a dict (for Redis/JSON storage)."""
         return {
@@ -113,6 +195,8 @@ def get_expired_sessions(self, idle_timeout_seconds: int) -> list[UptoSession]:
 
     def get_exhausted_sessions(self) -> list[UptoSession]: ...
 
+    def get_settlement_due_sessions(self, safety_margin_seconds: int) -> list[UptoSession]: ...
+
     @property
     def active_count(self) -> int: ...
 
@@ -271,6 +355,30 @@ def get_exhausted_sessions(self) -> list[UptoSession]:
                     exhausted.append(session)
         return exhausted
 
+    def get_settlement_due_sessions(self, safety_margin_seconds: int) -> list[UptoSession]:
+        """Get unsettled sessions whose signed authorization deadline is near.
+
+        These must be force-settled regardless of idle/exhaustion state, or the
+        on-chain settlement will revert as expired and the tab will be lost.
+
+        Args:
+            safety_margin_seconds: Settle this many seconds before the
+                authorization deadline to leave room for on-chain confirmation.
+
+        Returns:
+            List of sessions that should be settled now (deadline approaching).
+        """
+        now = time.time()
+        due: list[UptoSession] = []
+        with self._lock:
+            for session in self._sessions.values():
+                if session.settled or session.settling:
+                    continue
+                deadline = session.settlement_deadline
+                if deadline is not None and now >= deadline - safety_margin_seconds:
+                    due.append(session)
+        return due
+
     @property
     def active_count(self) -> int:
         """Number of active (unsettled) sessions."""
diff --git a/python/x402/tests/unit/http/middleware/test_flask.py b/python/x402/tests/unit/http/middleware/test_flask.py
index 044d066a12..8721e663f8 100644
--- a/python/x402/tests/unit/http/middleware/test_flask.py
+++ b/python/x402/tests/unit/http/middleware/test_flask.py
@@ -20,6 +20,7 @@
     PaymentMiddleware,
     ResponseWrapper,
     _check_if_bazaar_needed,
+    _session_key_from_payment,
     payment_middleware,
 )
 from x402.http.types import (
@@ -30,6 +31,7 @@
     RouteConfig,
 )
 from x402.schemas import PaymentPayload, PaymentRequirements
+from x402.session import SessionStore
 
 # =============================================================================
 # Helpers
@@ -760,3 +762,189 @@ def protected_route():
                 assert response.get_json() == {
                     "error": "Facilitator settle returned invalid data: {'success': true}"
                 }
+
+
+class TestSessionReuseGuards:
+    """Session reuse must treat an in-progress settlement as terminal."""
+
+    @staticmethod
+    def _make_session(store: SessionStore, *, deadline_offset: int = 3600) -> str:
+        import time
+
+        now = int(time.time())
+        return store.create_session(
+            permit_payload={
+                "payload": {"permit2Authorization": {"deadline": str(now + deadline_offset)}}
+            },
+            requirements={"network": "eip155:8453", "maxTimeoutSeconds": deadline_offset},
+            max_amount=1000,
+        )
+
+    def test_settling_session_is_not_reused(self):
+        """A session mid-settlement must return 402 (re-pay), never be reused.
+
+        Reusing a settling session would serve a response while add_cost()
+        rejects the charge — i.e. free inference. Guards the fix that adds
+        `session.settling` to the _handle_session_mode reuse check.
+        """
+        app = Flask(__name__)
+        store = SessionStore()
+        middleware = PaymentMiddleware(
+            app,
+            {},
+            MagicMock(),
+            sync_facilitator_on_start=False,
+            session_store=store,
+        )
+
+        payment_header = "test-payment-header"
+        session_id = self._make_session(store)
+        middleware._payment_to_session[_session_key_from_payment(payment_header)] = session_id
+
+        # Claim the session for settlement (as the reaper would).
+        assert store.mark_settling(session_id) is True
+
+        captured: dict[str, Any] = {}
+
+        def start_response(status: str, headers: list) -> None:
+            captured["status"] = status
+
+        body = b"".join(
+            middleware._handle_session_mode(
+                {}, start_response, MagicMock(), MagicMock(), payment_header
+            )
+        )
+
+        assert captured["status"].startswith("402")
+        assert b"expired or exhausted" in body
+
+    def test_active_session_is_reused(self):
+        """Control: an active (not settling) session is reused, not 402'd."""
+        app = Flask(__name__)
+        store = SessionStore()
+        middleware = PaymentMiddleware(
+            app,
+            {},
+            MagicMock(),
+            sync_facilitator_on_start=False,
+            session_store=store,
+        )
+
+        payment_header = "test-payment-header"
+        session_id = self._make_session(store)
+        middleware._payment_to_session[_session_key_from_payment(payment_header)] = session_id
+
+        captured: dict[str, Any] = {}
+
+        def start_response(status: str, headers: list) -> None:
+            captured["status"] = status
+
+        # Stub out the streaming path — we only care that reuse was allowed
+        # (i.e. the guard did NOT short-circuit to 402).
+        with patch.object(
+            middleware, "_stream_session_response", return_value=iter([b"ok"])
+        ) as streamed:
+            body = b"".join(
+                middleware._handle_session_mode(
+                    {}, start_response, MagicMock(), MagicMock(), payment_header
+                )
+            )
+
+        streamed.assert_called_once()
+        assert body == b"ok"
+        assert "status" not in captured  # no 402 emitted here
+
+
+class TestNewSessionAdmission:
+    """A fresh authorization may open a draw-down session only when it carries a
+    verifiable, not-near-expiry *signed* deadline. Admission gates paid work, so
+    it fails closed."""
+
+    @staticmethod
+    def _middleware(store: SessionStore) -> PaymentMiddleware:
+        app = Flask(__name__)
+        return PaymentMiddleware(
+            app,
+            {},
+            MagicMock(),
+            sync_facilitator_on_start=False,
+            session_store=store,
+        )
+
+    @staticmethod
+    def _result(payload: dict, *, amount: str = "1000", max_timeout: int = 3600) -> MagicMock:
+        result = MagicMock()
+        result.payment_payload.model_dump.return_value = payload
+        result.payment_requirements.model_dump.return_value = {
+            "network": "eip155:8453",
+            "maxTimeoutSeconds": max_timeout,
+        }
+        result.payment_requirements.amount = amount
+        return result
+
+    def _drive(self, middleware: PaymentMiddleware, result: MagicMock):
+        captured: dict[str, Any] = {}
+
+        def start_response(status: str, headers: list) -> None:
+            captured["status"] = status
+
+        with patch.object(
+            middleware, "_stream_session_response", return_value=iter([b"ok"])
+        ) as streamed:
+            body = b"".join(
+                middleware._handle_session_mode(
+                    {}, start_response, result, MagicMock(), "new-payment-header"
+                )
+            )
+        return captured, body, streamed
+
+    def test_rejects_session_without_signed_deadline(self):
+        """No parseable signed deadline → 402, and NO paid work is streamed.
+
+        The advertised-window fallback must not be trusted for admission: it can
+        outlive the true on-chain deadline, so settlement would revert and drop
+        the tab.
+        """
+        store = SessionStore()
+        middleware = self._middleware(store)
+        result = self._result({"payload": {"signature": "0x"}})  # no deadline
+
+        captured, body, streamed = self._drive(middleware, result)
+
+        assert captured["status"].startswith("402")
+        streamed.assert_not_called()
+        assert store.active_count == 0  # session never created
+        assert b"deadline" in body
+
+    def test_rejects_near_expiry_signed_deadline(self):
+        """A signed deadline inside the safety margin is refused (anti-churn)."""
+        import time
+
+        store = SessionStore()
+        middleware = self._middleware(store)  # default settlement_safety_margin=60
+        result = self._result(
+            {"payload": {"permit2Authorization": {"deadline": str(int(time.time()) + 30)}}}
+        )
+
+        captured, body, streamed = self._drive(middleware, result)
+
+        assert captured["status"].startswith("402")
+        streamed.assert_not_called()
+        assert store.active_count == 0
+
+    def test_admits_session_with_valid_signed_deadline(self):
+        """Control: a far-out signed deadline opens the session and serves."""
+        import time
+
+        store = SessionStore()
+        middleware = self._middleware(store)
+        result = self._result(
+            {"payload": {"permit2Authorization": {"deadline": str(int(time.time()) + 3600)}}}
+        )
+
+        captured, body, streamed = self._drive(middleware, result)
+
+        streamed.assert_called_once()
+        assert body == b"ok"
+        assert "status" not in captured
+        assert store.active_count == 1
diff --git a/python/x402/tests/unit/http/test_facilitator_client.py b/python/x402/tests/unit/http/test_facilitator_client.py
index 03d3ef5b58..95e948e2d7 100644
--- a/python/x402/tests/unit/http/test_facilitator_client.py
+++ b/python/x402/tests/unit/http/test_facilitator_client.py
@@ -115,3 +115,148 @@ def test_sync_settle_raises_facilitator_response_error_for_invalid_schema():
         match="Facilitator settle returned invalid data",
     ):
         client.settle(make_v2_payload(), make_payment_requirements())
+
+
+def _accepted_202_response() -> MagicMock:
+    """A facilitator 202 async-settlement acceptance carrying a job id."""
+    response = MagicMock(status_code=202, text='{"paymentJob": {"jobId": "payment-abc"}}')
+    response.json.return_value = {"paymentJob": {"jobId": "payment-abc"}}
+    return response
+
+
+def _job_status_response(body: dict) -> MagicMock:
+    response = MagicMock(status_code=200, text=json.dumps(body))
+    response.json.return_value = body
+    return response
+
+
+def test_sync_settle_202_optimistic_when_not_waiting():
+    """Default behavior: a 202 is reported as accepted (job id as transaction)."""
+    http_client = MagicMock()
+    http_client.post.return_value = _accepted_202_response()
+
+    client = HTTPFacilitatorClientSync(
+        FacilitatorConfig(url="https://facilitator.test", http_client=http_client)
+    )
+
+    result = client.settle(make_v2_payload(), make_payment_requirements())
+    assert result.success is True
+    assert result.transaction == "payment-abc"
+    http_client.get.assert_not_called()
+
+
+def test_sync_settle_waits_and_reports_onchain_success():
+    """With wait_for_settlement, poll the job and return the real tx on success."""
+    http_client = MagicMock()
+    http_client.post.return_value = _accepted_202_response()
+    http_client.get.side_effect = [
+        _job_status_response({"status": "processing"}),
+        _job_status_response(
+            {
+                "status": "succeeded",
+                "result": {
+                    "success": True,
+                    "transaction": "0xdeadbeef",
+                    "network": "eip155:8453",
+                },
+            }
+        ),
+    ]
+
+    client = HTTPFacilitatorClientSync(
+        FacilitatorConfig(
+            url="https://facilitator.test",
+            http_client=http_client,
+            wait_for_settlement=True,
+            settlement_poll_interval=0,
+        )
+    )
+
+    result = client.settle(make_v2_payload(), make_payment_requirements())
+    assert result.success is True
+    assert result.transaction == "0xdeadbeef"
+    assert http_client.get.call_count == 2
+
+
+def test_sync_settle_waits_and_reports_failure_when_job_reverts():
+    """Regression: a job that completes with success=False must NOT report success.
+
+    The facilitator marks a reverted/expired settlement job "succeeded" at the
+    queue level; the client must inspect result.success and surface a failure so
+    the session is not marked settled without an on-chain transaction.
+    """
+    http_client = MagicMock()
+    http_client.post.return_value = _accepted_202_response()
+    http_client.get.return_value = _job_status_response(
+        {
+            "status": "succeeded",
+            "result": {
+                "success": False,
+                "transaction": "",
+                "network": "eip155:8453",
+                "errorReason": "permit2_deadline_expired",
+            },
+        }
+    )
+
+    client = HTTPFacilitatorClientSync(
+        FacilitatorConfig(
+            url="https://facilitator.test",
+            http_client=http_client,
+            wait_for_settlement=True,
+            settlement_poll_interval=0,
+        )
+    )
+
+    result = client.settle(make_v2_payload(), make_payment_requirements())
+    assert result.success is False
+    assert result.error_reason == "permit2_deadline_expired"
+
+
+def test_sync_settle_waits_and_times_out_without_reporting_success():
+    """If the job never confirms, the client reports failure, not success."""
+    http_client = MagicMock()
+    http_client.post.return_value = _accepted_202_response()
+    http_client.get.return_value = _job_status_response({"status": "queued"})
+
+    client = HTTPFacilitatorClientSync(
+        FacilitatorConfig(
+            url="https://facilitator.test",
+            http_client=http_client,
+            wait_for_settlement=True,
+            settlement_poll_interval=0,
+            settlement_poll_timeout=0,
+        )
+    )
+
+    result = client.settle(make_v2_payload(), make_payment_requirements())
+    assert result.success is False
+    assert result.error_reason == "settlement_timeout"
+
+
+def test_negative_poll_values_are_clamped_dataclass_config():
+    """Negative poll interval/timeout must clamp to 0 (time.sleep rejects <0)."""
+    client = HTTPFacilitatorClientSync(
+        FacilitatorConfig(
+            url="https://facilitator.test",
+            wait_for_settlement=True,
+            settlement_poll_interval=-5.0,
+            settlement_poll_timeout=-1.0,
+        )
+    )
+    assert client._settlement_poll_interval == 0.0
+    assert client._settlement_poll_timeout == 0.0
+
+
+def test_negative_poll_values_are_clamped_dict_config():
+    """Same clamping applies when configured via a dict."""
+    client = HTTPFacilitatorClientSync(
+        {
+            "url": "https://facilitator.test",
+            "wait_for_settlement": True,
+            "settlement_poll_interval": -2.0,
+            "settlement_poll_timeout": -10.0,
+        }
+    )
+    assert client._settlement_poll_interval == 0.0
+    assert client._settlement_poll_timeout == 0.0
diff --git a/python/x402/tests/unit/test_session.py b/python/x402/tests/unit/test_session.py
new file mode 100644
index 0000000000..9c85d24b4f
--- /dev/null
+++ b/python/x402/tests/unit/test_session.py
@@ -0,0 +1,168 @@
+"""Unit tests for x402.session — upto draw-down session store.
+
+Focus: the settlement-deadline safeguards that prevent a long-lived session
+from outliving its signed authorization and silently dropping the tab.
+"""
+
+from __future__ import annotations
+
+import time
+
+from x402.session import (
+    SessionStore,
+    UptoSession,
+    signed_authorization_deadline,
+)
+
+
+def _permit_payload(deadline: int) -> dict:
+    """A minimal upto Permit2 payload carrying a signed deadline."""
+    return {
+        "x402Version": 2,
+        "scheme": "upto",
+        "network": "eip155:8453",
+        "payload": {
+            "permit2Authorization": {"deadline": str(deadline)},
+            "signature": "0xsig",
+        },
+    }
+
+
+def _requirements(max_timeout_seconds: int = 300) -> dict:
+    return {
+        "scheme": "upto",
+        "network": "eip155:8453",
+        "asset": "0x0000000000000000000000000000000000000000",
+        "amount": "1000",
+        "payTo": "0x1234567890123456789012345678901234567890",
+        "maxTimeoutSeconds": max_timeout_seconds,
+    }
+
+
+def test_settlement_deadline_prefers_signed_permit_deadline():
+    """The exact signed Permit2 deadline is used when present."""
+    deadline = int(time.time()) + 300
+    session = UptoSession(
+        session_id="s1",
+        permit_payload=_permit_payload(deadline),
+        requirements=_requirements(),
+        max_amount=1000,
+    )
+    assert session.settlement_deadline == float(deadline)
+
+
+def test_settlement_deadline_falls_back_to_created_plus_window():
+    """Without a signed deadline, fall back to created_at + maxTimeoutSeconds."""
+    session = UptoSession(
+        session_id="s2",
+        permit_payload={"payload": {"signature": "0xsig"}},  # no deadline
+        requirements=_requirements(max_timeout_seconds=450),
+        max_amount=1000,
+    )
+    assert session.settlement_deadline == session.created_at + 450.0
+
+
+def test_settlement_deadline_none_when_undeterminable():
+    session = UptoSession(
+        session_id="s3",
+        permit_payload={"payload": {}},
+        requirements={"scheme": "upto", "network": "eip155:8453"},
+        max_amount=1000,
+    )
+    assert session.settlement_deadline is None
+
+
+def test_signed_deadline_reads_permit2_deadline():
+    """signed_deadline reflects the canonical Permit2 signed deadline."""
+    deadline = int(time.time()) + 300
+    session = UptoSession(
+        session_id="s1",
+        permit_payload=_permit_payload(deadline),
+        requirements=_requirements(),
+        max_amount=1000,
+    )
+    assert session.signed_deadline == float(deadline)
+
+
+def test_signed_deadline_reads_eip3009_valid_before():
+    """EIP-3009 authorizations expose the deadline as validBefore."""
+    valid_before = int(time.time()) + 300
+    session = UptoSession(
+        session_id="s2",
+        permit_payload={"payload": {"authorization": {"validBefore": str(valid_before)}}},
+        requirements=_requirements(),
+        max_amount=1000,
+    )
+    assert session.signed_deadline == float(valid_before)
+
+
+def test_signed_deadline_is_none_without_signed_value():
+    """signed_deadline never substitutes the advertised-window fallback.
+
+    settlement_deadline may guess from created_at + maxTimeoutSeconds, but
+    signed_deadline must stay None so admission decisions can fail closed.
+    """
+    session = UptoSession(
+        session_id="s3",
+        permit_payload={"payload": {"signature": "0xsig"}},  # no deadline
+        requirements=_requirements(max_timeout_seconds=450),
+        max_amount=1000,
+    )
+    assert session.signed_deadline is None
+    # ...while settlement_deadline still falls back for the reaper.
+    assert session.settlement_deadline == session.created_at + 450.0
+
+
+def test_signed_authorization_deadline_helper_rejects_malformed():
+    """The extractor returns None (not a guess) for absent/garbled deadlines."""
+    assert signed_authorization_deadline({"payload": {}}) is None
+    assert signed_authorization_deadline({}) is None
+    assert (
+        signed_authorization_deadline(
+            {"payload": {"permit2Authorization": {"deadline": "not-a-number"}}}
+        )
+        is None
+    )
+
+
+def test_get_settlement_due_sessions_returns_near_deadline_sessions():
+    """A session inside the safety margin of its deadline is due for settlement."""
+    store = SessionStore()
+    now = int(time.time())
+
+    # Deadline 60s out — inside a 180s safety margin → due.
+    due_id = store.create_session(
+        permit_payload=_permit_payload(now + 60),
+        requirements=_requirements(),
+        max_amount=1000,
+    )
+    # Deadline 600s out — well outside the margin → not due.
+    fresh_id = store.create_session(
+        permit_payload=_permit_payload(now + 600),
+        requirements=_requirements(),
+        max_amount=1000,
+    )
+
+    due = store.get_settlement_due_sessions(safety_margin_seconds=180)
+    due_ids = {s.session_id for s in due}
+    assert due_id in due_ids
+    assert fresh_id not in due_ids
+
+
+def test_get_settlement_due_sessions_skips_settled_and_settling():
+    store = SessionStore()
+    now = int(time.time())
+    sid = store.create_session(
+        permit_payload=_permit_payload(now + 30),
+        requirements=_requirements(),
+        max_amount=1000,
+    )
+
+    assert store.get_settlement_due_sessions(180)  # due while active
+
+    store.mark_settling(sid)
+    assert store.get_settlement_due_sessions(180) == []  # claimed → skipped
+
+    store.clear_settling(sid)
+    store.mark_settled(sid, "0xabc")
+    assert store.get_settlement_due_sessions(180) == []  # settled → skipped