feat: add detect_rail_from_headers helper

agentscore_commerce/payment/__init__.py

@@ -9,7 +9,7 @@
     build_payment_request_blob,
     payment_directive,
 )
-from agentscore_commerce.payment.dispatch import dispatch_settlement_by_network
+from agentscore_commerce.payment.dispatch import detect_rail_from_headers, dispatch_settlement_by_network
 from agentscore_commerce.payment.headers import (
     BuildPaymentHeadersInput,
     PaymentHeadersRail,
@@ -124,6 +124,7 @@
     "coerce_resource_config",
     "create_mppx_server",
     "create_x402_server",
+    "detect_rail_from_headers",
     "dispatch_settlement_by_network",
     "extract_payment_signer",
     "extract_x402_signer",

agentscore_commerce/payment/dispatch.py

@@ -1,13 +1,44 @@
-"""Settlement dispatch by CAIP-2 network family (eip155→evm, solana→svm)."""
+"""Payment dispatch helpers.
+
+* :func:`detect_rail_from_headers` — detect which payment-protocol family
+  (x402 vs MPP) the inbound request carries, based on header presence.
+* :func:`dispatch_settlement_by_network` — route a settlement payload to
+  evm vs svm handler based on the CAIP-2 network family in
+  ``payload.accepted.network``.
+"""
 
 import inspect
-from collections.abc import Awaitable, Callable
-from typing import Any, TypeVar, cast
+from collections.abc import Awaitable, Callable, Mapping
+from typing import Any, Literal, TypeVar, cast
 
 T = TypeVar("T")
 Handler = Callable[[Any], T | Awaitable[T]]
 
 
+def detect_rail_from_headers(headers: Mapping[str, str]) -> Literal["x402", "mpp"] | None:
+    """Detect which payment-protocol family the inbound request carries.
+
+    Returns ``"mpp"`` when an ``Authorization`` header starts with the ``Payment``
+    scheme (case-insensitive per RFC 7235). Returns ``"x402"`` when a non-empty
+    ``payment-signature`` or ``x-payment`` header is present. Returns ``None``
+    otherwise.
+
+    In practice a client constructs a request with exactly one protocol's headers;
+    both arriving together is a client bug or misconfigured proxy. The helper
+    checks MPP first so the rare degenerate case resolves to MPP. Empty header
+    values are treated as absent. Header-name lookups are case-insensitive
+    (RFC 7230 §3.2). The narrower rail naming (``"tempo"`` vs ``"solana"`` inside
+    MPP) is merchant-side, derived from the credential body, not this helper.
+    """
+    lower = {k.lower(): v for k, v in headers.items()}
+    auth = lower.get("authorization") or ""
+    if auth.lower().startswith("payment "):
+        return "mpp"
+    if lower.get("payment-signature") or lower.get("x-payment"):
+        return "x402"
+    return None
+
+
 async def dispatch_settlement_by_network(
     payload: Any,
     *,

tests/test_dispatch.py

@@ -0,0 +1,64 @@
+"""Tests for ``agentscore_commerce.payment.dispatch.detect_rail_from_headers``.
+
+The fixture corpus below is locked as the cross-language contract with the
+Node sibling at ``node-commerce/tests/payment/detect_rail_from_headers.test.ts``.
+Both files reference identical header maps + expected results. A drift in either
+language (case-handling, empty-value treatment, scheme-prefix matching) fails
+that language's test against the locked value.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from agentscore_commerce.payment import detect_rail_from_headers
+
+# Cross-language fixtures: (label, headers_dict, expected_rail).
+_FIXTURES: list[tuple[str, dict[str, str], str | None]] = [
+    ("empty", {}, None),
+    ("payment_signature_only", {"payment-signature": "abc"}, "x402"),
+    ("x_payment_only", {"x-payment": "abc"}, "x402"),
+    ("authorization_payment", {"authorization": "Payment abc"}, "mpp"),
+    ("authorization_bearer", {"authorization": "Bearer xyz"}, None),
+    ("authorization_lowercase_scheme", {"authorization": "payment abc"}, "mpp"),
+    ("authorization_uppercase_name", {"Authorization": "Payment abc"}, "mpp"),
+    ("x_payment_uppercase_name", {"X-Payment": "abc"}, "x402"),
+    ("empty_values_dont_count", {"payment-signature": "", "x-payment": ""}, None),
+    (
+        "mpp_wins_when_both_present",
+        {"x-payment": "abc", "authorization": "Payment xyz"},
+        "mpp",
+    ),
+    ("payment_without_space_is_not_mpp", {"authorization": "PaymentNoSpace"}, None),
+    ("payment_with_only_space_is_mpp", {"authorization": "Payment "}, "mpp"),
+    ("mixed_case_authorization_name", {"AUTHORIZATION": "Payment abc"}, "mpp"),
+    ("authorization_uppercase_scheme", {"authorization": "PAYMENT abc"}, "mpp"),
+]
+
+
+@pytest.mark.parametrize(
+    ("label", "headers", "expected"),
+    _FIXTURES,
+    ids=[label for label, _, _ in _FIXTURES],
+)
+def test_locked_cross_language_fixture(
+    label: str,
+    headers: dict[str, str],
+    expected: str | None,
+) -> None:
+    del label  # `label` is consumed by parametrize ids; bind locally so linters don't flag it.
+    """Each fixture header set maps to the locked cross-language rail value."""
+    assert detect_rail_from_headers(headers) == expected
+
+
+def test_returns_x402_for_non_string_truthy_value() -> None:
+    """Any non-empty header value is treated as present (no validation of contents)."""
+    assert detect_rail_from_headers({"x-payment": "0"}) == "x402"
+
+
+def test_does_not_mutate_input_headers() -> None:
+    headers = {"X-Payment": "abc", "Authorization": "Payment xyz"}
+    detect_rail_from_headers(headers)
+    # Keys preserved verbatim; helper only reads via a lowercase projection.
+    assert "X-Payment" in headers
+    assert "Authorization" in headers