feat: add zero_amount_carve_out helper

agentscore_commerce/payment/__init__.py

@@ -77,6 +77,11 @@
     validate_x402_network_config,
     verify_x402_request,
 )
+from agentscore_commerce.payment.zero_settle import (
+    ZeroSettleRail,
+    ZeroSettleResult,
+    zero_amount_carve_out,
+)
 
 __all__ = [
     "SETTLEMENT_OVERRIDES_HEADER",
@@ -113,6 +118,8 @@
     "X402AcceptsBlock",
     "X402FacilitatorChoice",
     "X402SymbolicRail",
+    "ZeroSettleRail",
+    "ZeroSettleResult",
     "alias_amount_fields",
     "build_idempotency_key",
     "build_payment_directive",
@@ -143,4 +150,5 @@
     "validate_x402_network_config",
     "verify_x402_request",
     "www_authenticate_header",
+    "zero_amount_carve_out",
 ]

agentscore_commerce/payment/zero_settle.py

@@ -0,0 +1,94 @@
+"""Zero-amount carve-out: skip upstream verify+settle for $0 orders.
+
+CDP rejects EIP-3009 ``transferWithAuthorization`` with ``value=0`` as
+``invalid_payload``; pympp's tempo intents accept only ``hash`` and
+``transaction`` payload types (rejecting the ``proof`` payload that ``mppx``
+emits for $0 settles). Both upstream verify+settle paths fail when the
+authorized amount is zero, so merchants that drop the settle to $0 in a
+redemption-code flow need a way to skip verify+settle entirely while still
+recovering the signer for wallet-capture attribution.
+
+``zero_amount_carve_out`` is that path: parse the credential, lift the signer,
+return ``ZeroSettleResult(signer_address, signer_network, tx_hash=None)``.
+Identity is still authenticated by the merchant's gate above; the redemption
+code is single-use; nothing on-chain to verify.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Literal
+
+from agentscore_commerce.identity.address import is_valid_evm_address, normalize_address
+from agentscore_commerce.payment.signer import SignerNetwork, extract_payment_signer
+
+ZeroSettleRail = Literal["x402-base", "tempo", "solana"]
+
+
+@dataclass(frozen=True)
+class ZeroSettleResult:
+    """Result of a zero-amount carve-out: signer info + always-null tx hash.
+
+    ``tx_hash`` is intentionally fixed to ``None``: a zero-amount carve-out
+    skips on-chain settlement entirely, so no transaction hash exists. The
+    field is present so callers can use ``ZeroSettleResult`` interchangeably
+    with the success path of ``process_x402_settle`` etc. without branching.
+    """
+
+    signer_address: str | None
+    signer_network: SignerNetwork | None
+    tx_hash: None = None
+
+
+def zero_amount_carve_out(
+    *,
+    rail: ZeroSettleRail,
+    payload: dict[str, Any] | None = None,
+    authorization_header: str | None = None,
+) -> ZeroSettleResult:
+    """Skip verify+settle for a zero-amount order; recover the signer from the credential.
+
+    For ``rail="x402-base"``: pass ``payload`` (the verified x402 dict, typically
+    ``verify_x402_request(...).payload``). Reads
+    ``payload["payload"]["authorization"]["from"]``.
+
+    For ``rail="tempo"`` or ``"solana"``: pass ``authorization_header`` (the full
+    ``Authorization: Payment <base64>`` header value). Reads the ``did:pkh:*``
+    source DID via :func:`extract_payment_signer`.
+
+    Returns :class:`ZeroSettleResult`. ``signer_address`` and ``signer_network``
+    are ``None`` when the credential is malformed, missing required fields,
+    or shaped wrong for the requested rail. ``tx_hash`` is always ``None``
+    since no on-chain settle runs.
+    """
+    if rail == "x402-base":
+        return _x402_signer_from_payload(payload)
+    if rail in ("tempo", "solana"):
+        return _mpp_signer_from_auth(authorization_header)
+    return ZeroSettleResult(signer_address=None, signer_network=None)
+
+
+def _x402_signer_from_payload(payload: dict[str, Any] | None) -> ZeroSettleResult:
+    """Read the EVM signer from a verified x402 EIP-3009 payload dict."""
+    if not isinstance(payload, dict):
+        return ZeroSettleResult(signer_address=None, signer_network=None)
+    inner = payload.get("payload")
+    if not isinstance(inner, dict):
+        return ZeroSettleResult(signer_address=None, signer_network=None)
+    authorization = inner.get("authorization")
+    if not isinstance(authorization, dict):
+        return ZeroSettleResult(signer_address=None, signer_network=None)
+    from_addr = authorization.get("from")
+    if not isinstance(from_addr, str) or not is_valid_evm_address(from_addr):
+        return ZeroSettleResult(signer_address=None, signer_network=None)
+    return ZeroSettleResult(signer_address=normalize_address(from_addr), signer_network="evm")
+
+
+def _mpp_signer_from_auth(authorization_header: str | None) -> ZeroSettleResult:
+    """Read the signer from an MPP ``Authorization: Payment <base64>`` header."""
+    if not isinstance(authorization_header, str):
+        return ZeroSettleResult(signer_address=None, signer_network=None)
+    signer = extract_payment_signer(authorization_header=authorization_header)
+    if signer is None:
+        return ZeroSettleResult(signer_address=None, signer_network=None)
+    return ZeroSettleResult(signer_address=signer.address, signer_network=signer.network)

tests/test_zero_settle.py

@@ -0,0 +1,175 @@
+"""Tests for ``agentscore_commerce.payment.zero_settle.zero_amount_carve_out``.
+
+Locked cross-language fixtures shared with the Node sibling at
+``node-commerce/tests/payment/zero_settle.test.ts``. Both files reference
+identical payload dicts / Authorization header values + expected
+``ZeroSettleResult``. Drift in either language (DID parsing, dict-shape
+handling, address validation) fails that language's test against the
+locked value.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from agentscore_commerce.payment import ZeroSettleResult, zero_amount_carve_out
+
+# ─── x402-base rail: payload is the verified outer dict (already base64-decoded) ────────
+
+_X402_EVM_PAYLOAD = {
+    "payload": {"authorization": {"from": "0xABCDef1234567890123456789012345678901234"}},
+}
+
+_X402_FIXTURES = [
+    (
+        "x402_evm_signer_recovered",
+        _X402_EVM_PAYLOAD,
+        ZeroSettleResult(
+            signer_address="0xabcdef1234567890123456789012345678901234",
+            signer_network="evm",
+        ),
+    ),
+    ("x402_payload_none", None, ZeroSettleResult(signer_address=None, signer_network=None)),
+    (
+        "x402_payload_not_dict",
+        "not-a-dict",  # type: ignore[arg-type]
+        ZeroSettleResult(signer_address=None, signer_network=None),
+    ),
+    (
+        "x402_inner_payload_missing",
+        {},
+        ZeroSettleResult(signer_address=None, signer_network=None),
+    ),
+    (
+        "x402_inner_payload_not_dict",
+        {"payload": "oops"},
+        ZeroSettleResult(signer_address=None, signer_network=None),
+    ),
+    (
+        "x402_authorization_missing",
+        {"payload": {}},
+        ZeroSettleResult(signer_address=None, signer_network=None),
+    ),
+    (
+        "x402_from_missing",
+        {"payload": {"authorization": {}}},
+        ZeroSettleResult(signer_address=None, signer_network=None),
+    ),
+    (
+        "x402_from_not_evm_shape",
+        {"payload": {"authorization": {"from": "not-an-address"}}},
+        ZeroSettleResult(signer_address=None, signer_network=None),
+    ),
+]
+
+# ─── tempo / solana MPP rails: authorization_header carries the credential ──────────────
+
+_MPP_TEMPO_AUTH = (
+    "Payment eyJzb3VyY2UiOiAiZGlkOnBraDplaXAxNTU6NDIxNzoweEFCQ0RlZjEyMzQ1Njc4OTAxMjM0NTY3ODkwMTIzNDU2Nzg5MDEyMzQifQ=="
+)
+_MPP_SOLANA_AUTH = (
+    "Payment "
+    "eyJjaGFsbGVuZ2UiOiB7InNvdXJjZSI6ICJkaWQ6cGtoOnNvbGFuYTo1ZXlrdDRVc0Z2OFA4TkpkVFJFcFkxdnpxS3FaS3ZkcFVrZkZw"
+    "OjduUUVneHFFVzFiRHFhVDNrWldhOEtxVWs0V2ZoNFZiY3cifX0="
+)
+
+_MPP_FIXTURES = [
+    (
+        "mpp_tempo_signer_recovered",
+        "tempo",
+        _MPP_TEMPO_AUTH,
+        ZeroSettleResult(
+            signer_address="0xabcdef1234567890123456789012345678901234",
+            signer_network="evm",
+        ),
+    ),
+    (
+        "mpp_solana_signer_recovered",
+        "solana",
+        _MPP_SOLANA_AUTH,
+        ZeroSettleResult(
+            signer_address="7nQEgxqEW1bDqaT3kZWa8KqUk4Wfh4Vbcw",
+            signer_network="solana",
+        ),
+    ),
+    ("mpp_auth_none", "tempo", None, ZeroSettleResult(signer_address=None, signer_network=None)),
+    ("mpp_auth_empty", "tempo", "", ZeroSettleResult(signer_address=None, signer_network=None)),
+    (
+        "mpp_auth_not_payment_scheme",
+        "tempo",
+        "Bearer abc.def.ghi",
+        ZeroSettleResult(signer_address=None, signer_network=None),
+    ),
+    (
+        "mpp_credential_without_source",
+        "tempo",
+        "Payment eyJmb28iOiAiYmFyIn0=",  # {"foo": "bar"} — no source field
+        ZeroSettleResult(signer_address=None, signer_network=None),
+    ),
+]
+
+
+@pytest.mark.parametrize(
+    ("label", "payload", "expected"),
+    _X402_FIXTURES,
+    ids=[label for label, _, _ in _X402_FIXTURES],
+)
+def test_x402_base_locked_fixture(label, payload, expected) -> None:
+    del label
+    assert zero_amount_carve_out(rail="x402-base", payload=payload) == expected
+
+
+@pytest.mark.parametrize(
+    ("label", "rail", "auth_header", "expected"),
+    _MPP_FIXTURES,
+    ids=[label for label, _, _, _ in _MPP_FIXTURES],
+)
+def test_mpp_locked_fixture(label, rail, auth_header, expected) -> None:
+    del label
+    assert zero_amount_carve_out(rail=rail, authorization_header=auth_header) == expected
+
+
+def test_tx_hash_is_always_none() -> None:
+    """The carve-out skips on-chain settle; tx_hash is fixed to None."""
+    result = zero_amount_carve_out(rail="x402-base", payload=_X402_EVM_PAYLOAD)
+    assert result.tx_hash is None
+
+
+def test_x402_base_ignores_authorization_header() -> None:
+    """``rail="x402-base"`` only reads ``payload``; an extra ``authorization_header`` arg is ignored."""
+    result = zero_amount_carve_out(
+        rail="x402-base",
+        payload=_X402_EVM_PAYLOAD,
+        authorization_header=_MPP_SOLANA_AUTH,
+    )
+    # Returns the x402 EVM signer, NOT the Solana signer from the MPP header
+    assert result.signer_address == "0xabcdef1234567890123456789012345678901234"
+    assert result.signer_network == "evm"
+
+
+def test_mpp_rails_ignore_payload() -> None:
+    """``rail="tempo"`` / ``"solana"`` only read ``authorization_header``; ``payload`` is ignored."""
+    result = zero_amount_carve_out(
+        rail="tempo",
+        payload=_X402_EVM_PAYLOAD,
+        authorization_header=_MPP_TEMPO_AUTH,
+    )
+    # Returns the tempo MPP signer from the authorization header, not the x402 payload's from
+    assert result.signer_address == "0xabcdef1234567890123456789012345678901234"
+    assert result.signer_network == "evm"
+
+
+def test_no_credential_provided_returns_none() -> None:
+    """Missing both ``payload`` and ``authorization_header`` returns a null result."""
+    assert zero_amount_carve_out(rail="x402-base") == ZeroSettleResult(
+        signer_address=None,
+        signer_network=None,
+    )
+    assert zero_amount_carve_out(rail="tempo") == ZeroSettleResult(
+        signer_address=None,
+        signer_network=None,
+    )
+    assert zero_amount_carve_out(rail="solana") == ZeroSettleResult(
+        signer_address=None,
+        signer_network=None,
+    )