feat: collapsed signer_match + per-adapter quota/fail-open helpers (python parity)

CLAUDE.md

@@ -17,7 +17,7 @@ Every helper is extracted from a real consumer, not speculated.
 
 ## Architecture
 
-Single Python package, hatchling-built, published to PyPI as `agentscore-commerce`. Per-framework identity adapters expose the same surface — `AgentScoreGate` (or `agentscore_gate(app, ...)` for Flask/Sanic), `capture_wallet`, `verify_wallet_signer_match`, `get_assess_data` — with network-aware address normalization (EVM lowercased, Solana base58 preserved verbatim).
+Single Python package, hatchling-built, published to PyPI as `agentscore-commerce`. Per-framework identity adapters expose the same surface — `AgentScoreGate` (or `agentscore_gate(app, ...)` for Flask/Sanic), `capture_wallet`, `verify_wallet_signer_match`, `get_assess_data`, `get_gate_degraded_state`, `get_gate_quota_info` — with network-aware address normalization (EVM lowercased, Solana base58 preserved verbatim).
 
 | Directory | Contents |
 |---|---|
@@ -56,6 +56,12 @@ Two identity types: wallet (`X-Wallet-Address`) and operator-token (`X-Operator-
 
 Captured wallets: `capture_wallet(...)` is fire-and-forget — reads `operator_token` stashed during gating and POSTs to `/v1/credentials/wallets`. No-ops for wallet-authenticated requests.
 
+Wallet-signer-match: `verify_wallet_signer_match` / `averify_wallet_signer_match` makes a single `/v1/assess` call with `resolve_signer` set; the API resolves both wallets and emits a `signer_match` verdict in the same response — collapses the legacy 2 follow-up assess calls into one round trip. Repeat lookups for the same `(claimed, signer)` pair hit a per-cache-entry `signer_match_by_signer` sub-dict and skip the API entirely. Falls back to a 2-resolve path when the API doesn't emit `signer_match` (canary rollout safety).
+
+### Fail-open (opt-in)
+
+`fail_open=True` on `AgentScoreGate(...)` (or `agentscore_gate(app, ...)`) flips infra-failure handling: 429 / 5xx / network-timeout pass through to the handler with the gate state stamped `degraded=True` + `infra_reason="quota_exceeded" | "api_error" | "network_timeout"`. `get_gate_degraded_state(request)` (Flask: `get_gate_degraded_state()` — reads from `g`) returns `{"degraded": bool, "infra_reason"?: str}` for merchant logging/alerting. Default stays `fail_open=False` — regulated commerce should keep it. Compliance denials (sanctions, age, jurisdiction, signer-mismatch) still deny regardless of the flag. The gate's `try` wraps only the AgentScore call — never the downstream user handler.
+
 ### Mount posture: gate-first vs gate-conditional
 
 `AgentScoreGate(...)` (or `agentscore_gate(app, ...)` on Flask/Sanic) is mounted directly when the route is AgentScore-only — every request runs identity + policy. To support **anonymous discovery by any spec-compliant x402 wallet** (Coinbase awal, Phantom, Solflare, …), wrap the gate so it fires only when a payment credential is attached:

README.md

@@ -292,6 +292,32 @@ async def purchase(request: Request):
     return JSONResponse(result.body, status_code=result.status, headers=result.headers)
 ```
 
+## Fail-open behavior
+
+By default AgentScore Gate fails closed: any AgentScore-side infrastructure failure (HTTP 429, 5xx, network timeout) returns 503 to the buyer. Set `fail_open=True` on `AgentScoreGate(...)` to opt in to graceful degradation:
+
+```python
+from fastapi import Depends, FastAPI, Request
+from agentscore_commerce.identity.fastapi import AgentScoreGate, get_gate_degraded_state
+
+app = FastAPI()
+gate = AgentScoreGate(api_key=os.environ["AGENTSCORE_API_KEY"], fail_open=True)
+
+@app.post("/purchase", dependencies=[Depends(gate)])
+async def purchase(request: Request):
+    state = get_gate_degraded_state(request)
+    if state["degraded"]:
+        # Compliance was NOT enforced this request — log/alert/refund-async/etc.
+        logger.warning("gate degraded: %s", state["infra_reason"])
+    # ...rest of handler
+```
+
+When `fail_open=True` AND the failure is infra-shape, the gate state carries `degraded=True` + `infra_reason="quota_exceeded" | "api_error" | "network_timeout"` so merchants can log/alert without parsing console output. **Compliance denials (sanctions, age, jurisdiction, signer-mismatch) still deny regardless of `fail_open`** — `fail_open` only covers "AgentScore couldn't tell us," never "AgentScore said no."
+
+For regulated commerce (alcohol, age-gated, sanctioned-jurisdiction-relevant) keep the default `fail_open=False` — outage is the correct posture; bypassing compliance on infra failure is a compliance gap. For low-stakes commerce or high-uptime SLAs, opt in and use the `degraded` flag as the audit trail.
+
+The `get_gate_degraded_state` helper is exported by every framework adapter (FastAPI, Flask, Django, AIOHTTP, Sanic, ASGI middleware) and reads from the framework-appropriate request state. The signature takes a request argument everywhere except Flask, which reads from `g` and takes no arguments.
+
 ## Examples
 
 The [examples/](./examples) directory has 7 runnable single-file FastAPI apps covering common merchant scenarios. See [examples/README.md](./examples/README.md) for the full table.

agentscore_commerce/identity/_response.py

@@ -164,20 +164,19 @@
         "action": "contact_merchant",
         "steps": [
             (
-                "The merchant's AgentScore tier does not include the assess feature, so "
-                "agent identity cannot be evaluated. This is a merchant-side configuration "
-                "gap — there is no agent-side recovery."
+                "The merchant's AgentScore account does not have the assess endpoint "
+                "enabled, so agent identity cannot be evaluated. This is a merchant-side "
+                "configuration gap — there is no agent-side recovery."
             ),
             (
                 "Contact the merchant (their support channel — typically listed in "
-                "/llms.txt or the OpenAPI servers metadata) and request they upgrade "
-                "their AgentScore plan."
+                "/llms.txt or the OpenAPI servers metadata) so they can resolve the "
+                "configuration on their side."
             ),
         ],
         "user_message": (
-            "This merchant's identity gate is misconfigured (AgentScore tier doesn't "
-            "support assess). Contact the merchant — there's nothing to fix on the "
-            "agent side."
+            "This merchant's identity gate is misconfigured. Contact the merchant — "
+            "there's nothing to fix on the agent side."
         ),
     }
 )
@@ -224,10 +223,46 @@
     }
 )
 
+_API_ERROR_INSTRUCTIONS = json.dumps(
+    {
+        "action": "retry_with_backoff",
+        "steps": [
+            "Verification is temporarily unavailable. Retry the request after 5-30 seconds with exponential backoff.",
+            "This is NOT a compliance denial — the user does not need to re-verify their "
+            "identity. Send the same identity headers (X-Wallet-Address or X-Operator-Token) "
+            "on retry.",
+            "If the request continues to fail after 3+ retries (~60 seconds total), surface the "
+            "error to the user with the merchant's support contact.",
+        ],
+        "user_message": (
+            "Verification is temporarily unavailable. Please try again in a moment — this is a "
+            "transient issue, not a problem with your account."
+        ),
+    }
+)
+
+QUOTA_EXCEEDED_INSTRUCTIONS = json.dumps(
+    {
+        "action": "contact_merchant",
+        "steps": [
+            "AgentScore identity verification is unavailable for this merchant. This is a "
+            "merchant-side issue and is NOT recoverable via retry.",
+            "Do not retry: the same 503 will be returned until the merchant resolves the issue on their side.",
+            "Surface to the user with the merchant's support contact. The merchant (not the agent) needs to act.",
+        ],
+        "user_message": (
+            "This merchant's identity verification is temporarily unavailable. Try again later, "
+            "or contact the merchant directly."
+        ),
+    }
+)
+
+
 # Default agent_instructions per denial code. Adapters can override by passing
 # ``agent_instructions=`` on the DenialReason; otherwise the body emitter looks
 # up this map so every denial carries a machine-readable next step.
 _DEFAULT_AGENT_INSTRUCTIONS: dict[str, str] = {
+    "api_error": _API_ERROR_INSTRUCTIONS,
     "missing_identity": _MISSING_IDENTITY_INSTRUCTIONS,
     "wallet_signer_mismatch": WALLET_SIGNER_MISMATCH_INSTRUCTIONS,
     "wallet_auth_requires_wallet_signing": WALLET_AUTH_REQUIRES_WALLET_SIGNING_INSTRUCTIONS,
@@ -260,7 +295,7 @@ def build_missing_identity_reason() -> DenialReason:
     ),
     "wallet_not_trusted": "The wallet does not meet the merchant compliance policy.",
     "api_error": "AgentScore is unreachable. This is transient — retry in a few seconds.",
-    "payment_required": "AgentScore tier does not support assess. Contact support.",
+    "payment_required": "Assess endpoint not enabled for this merchant. Contact support.",
     "wallet_signer_mismatch": (
         "Payment signer does not match the wallet claimed via X-Wallet-Address. The signer and the "
         "claimed wallet must both resolve to the same AgentScore operator."
@@ -321,10 +356,6 @@ def denial_reason_to_body(reason: DenialReason) -> dict[str, Any]:
         body["actual_signer"] = reason.actual_signer
     if reason.linked_wallets:
         body["linked_wallets"] = reason.linked_wallets
-    # api_error denials get a default retry hint so agents know it's transient. Vendors can
-    # override by spreading their own next_steps into a custom on_denied body.
-    if reason.code == "api_error" and not (reason.extra and reason.extra.get("next_steps")):
-        body["next_steps"] = {"action": "retry", "retry_after_seconds": 5}
     # Merchant-supplied fields from on_before_session hook. Guard against collision
     # with reserved fields — the gate owns those and can't let a hook override them.
     if reason.extra:

agentscore_commerce/identity/aiohttp.py

@@ -4,6 +4,8 @@
 
 from typing import TYPE_CHECKING, Any
 
+import httpx
+
 from agentscore_commerce.identity._denial import (
     FIXABLE_DENIAL_REASONS,
     build_contact_support_next_steps,
@@ -12,11 +14,16 @@
     is_fixable_denial,
     verification_agent_instructions,
 )
-from agentscore_commerce.identity._response import build_missing_identity_reason, denial_reason_to_body
+from agentscore_commerce.identity._response import (
+    QUOTA_EXCEEDED_INSTRUCTIONS,
+    build_missing_identity_reason,
+    denial_reason_to_body,
+)
 from agentscore_commerce.identity.client import (
     GateClient,
     InvalidCredentialError,
     PaymentRequiredError,
+    QuotaExceededError,
     TokenDeniedError,
     build_invalid_credential_reason,
     build_token_denied_reason,
@@ -25,9 +32,11 @@
 from agentscore_commerce.identity.types import (
     AgentIdentity,
     DenialReason,
+    GateQuotaInfo,
     Network,
     VerifyWalletSignerMatchOptions,
     VerifyWalletSignerResult,
+    apply_degraded,
 )
 from agentscore_commerce.payment.signer import (
     extract_payment_signer,
@@ -45,6 +54,12 @@
 GATE_STATE_KEY = "__agentscore_gate"
 ASSESS_STATE_KEY = "agentscore"
 
+
+def _mark_degraded_aiohttp(request: web.Request, infra_reason: str) -> None:
+    """Stamp the gate state on an aiohttp request as fail-open'd."""
+    apply_degraded(request.get(GATE_STATE_KEY), infra_reason)
+
+
 __all__ = [
     "FIXABLE_DENIAL_REASONS",
     "CreateSessionOnMissing",
@@ -57,6 +72,8 @@
     "extract_payment_signer",
     "extract_payment_signer_address",
     "get_assess_data",
+    "get_gate_degraded_state",
+    "get_gate_quota_info",
     "is_fixable_denial",
     "read_x402_payment_header",
     "verification_agent_instructions",
@@ -73,6 +90,31 @@ def get_assess_data(request: web.Request) -> dict[str, Any] | None:
     return request.get(ASSESS_STATE_KEY)
 
 
+def get_gate_degraded_state(request: web.Request) -> dict[str, Any]:
+    """Return whether the gate fail-open'd due to AgentScore-side infra failure.
+
+    Returns ``{"degraded": False}`` for normal allows; ``{"degraded": True,
+    "infra_reason": "quota_exceeded" | "api_error" | "network_timeout"}`` when bypassed.
+    """
+    state = request.get(GATE_STATE_KEY)
+    if isinstance(state, dict) and state.get("degraded"):
+        return {"degraded": True, "infra_reason": state.get("infra_reason")}
+    return {"degraded": False}
+
+
+def get_gate_quota_info(request: web.Request) -> GateQuotaInfo | None:
+    """Read AgentScore assess quota observability for this request.
+
+    Captured from ``X-Quota-*`` response headers on this request's gate evaluate.
+    """
+    state = request.get(GATE_STATE_KEY)
+    if isinstance(state, dict):
+        quota = state.get("quota")
+        if isinstance(quota, GateQuotaInfo):
+            return quota
+    return None
+
+
 def _default_extract_identity(request: web.Request) -> AgentIdentity | None:
     token = request.headers.get(DEFAULT_TOKEN_HEADER)
     addr = request.headers.get(DEFAULT_ADDRESS_HEADER)
@@ -172,39 +214,11 @@ async def _agentscore_middleware(
 
         chain_override = _extract_chain(request)
 
+        # Only acheck_identity is wrapped — the downstream handler call must NOT be in the
+        # try, otherwise an exception in the user's route would be misclassified as an
+        # AgentScore infra failure and (under fail_open) re-invoke their handler.
         try:
             result = await client.acheck_identity(identity, chain_override)
-
-            if result.allow:
-                request["agentscore"] = result.raw
-                return await handler(request)
-
-            # Fixable compliance denials (kyc_required, kyc_pending, kyc_failed) get the
-            # same UX as missing_identity: the gate mints a fresh verification session,
-            # the agent polls until status=verified, gets a fresh opc_..., and retries
-            # with X-Operator-Token. Unfixable reasons (sanctions_flagged, age_insufficient,
-            # jurisdiction_restricted) keep the bare wallet_not_trusted denial.
-            # `jurisdiction_restricted` is unfixable: the API only emits it after KYC is
-            # verified (the user's KYC'd country is in the blocked list — re-doing KYC
-            # won't change the country).
-            if is_fixable_denial(result.reasons) and create_session_on_missing is not None:
-                session_reason = await try_create_session_denial_reason(
-                    create_session_on_missing,
-                    client.user_agent,
-                    request,
-                )
-                if session_reason is not None:
-                    body, status = _on_denied(request, session_reason)
-                    return web.json_response(body, status=status)
-
-            reason = DenialReason(
-                code="wallet_not_trusted",
-                decision=result.decision,
-                reasons=result.reasons,
-                verify_url=result.verify_url,
-            )
-            body, status = _on_denied(request, reason)
-            return web.json_response(body, status=status)
         except PaymentRequiredError:
             if client.fail_open:
                 return await handler(request)
@@ -218,12 +232,63 @@ async def _agentscore_middleware(
             # Permanent — no auto-session, agent should switch tokens or restart.
             body, status = _on_denied(request, build_invalid_credential_reason())
             return web.json_response(body, status=status)
+        except QuotaExceededError:
+            if client.fail_open:
+                _mark_degraded_aiohttp(request, "quota_exceeded")
+                return await handler(request)
+            body, status = _on_denied(
+                request,
+                DenialReason(code="api_error", agent_instructions=QUOTA_EXCEEDED_INSTRUCTIONS),
+            )
+            return web.json_response(body, status=status)
+        except httpx.TimeoutException:
+            if client.fail_open:
+                _mark_degraded_aiohttp(request, "network_timeout")
+                return await handler(request)
+            body, status = _on_denied(request, DenialReason(code="api_error"))
+            return web.json_response(body, status=status)
         except Exception:
             if client.fail_open:
+                _mark_degraded_aiohttp(request, "api_error")
                 return await handler(request)
             body, status = _on_denied(request, DenialReason(code="api_error"))
             return web.json_response(body, status=status)
 
+        if result.allow:
+            request["agentscore"] = result.raw
+            if result.quota is not None:
+                state = request.get(GATE_STATE_KEY)
+                if isinstance(state, dict):
+                    state["quota"] = result.quota
+            return await handler(request)
+
+        # Fixable compliance denials (kyc_required, kyc_pending, kyc_failed) get the
+        # same UX as missing_identity: the gate mints a fresh verification session,
+        # the agent polls until status=verified, gets a fresh opc_..., and retries
+        # with X-Operator-Token. Unfixable reasons (sanctions_flagged, age_insufficient,
+        # jurisdiction_restricted) keep the bare wallet_not_trusted denial.
+        # `jurisdiction_restricted` is unfixable: the API only emits it after KYC is
+        # verified (the user's KYC'd country is in the blocked list — re-doing KYC
+        # won't change the country).
+        if is_fixable_denial(result.reasons) and create_session_on_missing is not None:
+            session_reason = await try_create_session_denial_reason(
+                create_session_on_missing,
+                client.user_agent,
+                request,
+            )
+            if session_reason is not None:
+                body, status = _on_denied(request, session_reason)
+                return web.json_response(body, status=status)
+
+        reason = DenialReason(
+            code="wallet_not_trusted",
+            decision=result.decision,
+            reasons=result.reasons,
+            verify_url=result.verify_url,
+        )
+        body, status = _on_denied(request, reason)
+        return web.json_response(body, status=status)
+
     return _agentscore_middleware