feat: add classify_orchestration_error helper

agentscore_commerce/payment/__init__.py

@@ -61,6 +61,7 @@
     ProcessX402SettleInput,
     ProcessX402SettleResult,
     ProcessX402SettleSuccess,
+    classify_orchestration_error,
     classify_x402_settle_result,
     coerce_payment_payload,
     coerce_resource_config,
@@ -126,6 +127,7 @@
     "build_payment_headers",
     "build_payment_request_blob",
     "build_x402_accepts_for_402",
+    "classify_orchestration_error",
     "classify_x402_settle_result",
     "coerce_payment_payload",
     "coerce_resource_config",

agentscore_commerce/payment/x402_settle.py

@@ -189,6 +189,79 @@ def classify_x402_settle_result(result: ProcessX402SettleResult) -> ClassifiedX4
     return None
 
 
+def classify_orchestration_error(err: BaseException | str) -> ClassifiedX402Error | None:
+    """Classify a thrown error during the 402 orchestration.
+
+    Catches errors that escape ``process_x402_settle`` (e.g. raised by ``mppx.compose``,
+    a Stripe SDK call, or any other payment-side library code wrapped in a single
+    ``try/except`` around the full settle flow). Returns a :class:`ClassifiedX402Error`
+    when the error message matches a known pattern; ``None`` otherwise.
+
+    Callers should rethrow on ``None`` — this helper never swallows unknown errors.
+    The typical pattern::
+
+        try:
+            ...
+        except Exception as exc:
+            classified = classify_orchestration_error(exc)
+            if classified is not None:
+                return JSONResponse(
+                    {"error": {"code": classified.code, "message": classified.message},
+                     "next_steps": classified.next_steps},
+                    status_code=classified.status,
+                )
+            log.error("unclassified payment error: %s", exc)
+            raise
+
+    Pattern matching is case-insensitive substring on the error message:
+
+    * ``"x402version"`` / ``"invalid payment"`` / ``"unsupported x402"`` →
+      400 ``payment_proof_invalid`` / ``regenerate_payment_credential``
+    * ``"stripe"`` / ``"facilitator"`` / ``"cdp"`` →
+      503 ``payment_provider_unavailable`` / ``retry_or_swap_method``
+    * Anything else → ``None`` (caller rethrows)
+
+    Substring matching is intentionally narrow. New error families should land here
+    explicitly rather than have the helper grow opaque heuristics. For tagged failure
+    results that already classify themselves, use :func:`classify_x402_settle_result`.
+    """
+    msg = str(err) if isinstance(err, BaseException) else err
+    if not isinstance(msg, str):
+        return None
+    msg_lower = msg.lower()
+
+    if any(needle in msg_lower for needle in ("x402version", "invalid payment", "unsupported x402")):
+        return ClassifiedX402Error(
+            status=400,
+            code="payment_proof_invalid",
+            message="Payment credential is malformed or uses an unsupported version",
+            next_steps={
+                "action": "regenerate_payment_credential",
+                "user_message": (
+                    "The payment credential is malformed or uses an unsupported version. "
+                    "Regenerate from a fresh 402 challenge and re-sign."
+                ),
+            },
+        )
+
+    if any(needle in msg_lower for needle in ("stripe", "facilitator", "cdp")):
+        return ClassifiedX402Error(
+            status=503,
+            code="payment_provider_unavailable",
+            message="Payment provider returned an error",
+            next_steps={
+                "action": "retry_or_swap_method",
+                "retry_after_seconds": 10,
+                "user_message": (
+                    "Transient payment-provider error. Retry in a few seconds, "
+                    "or pick a different rail from the 402 challenge."
+                ),
+            },
+        )
+
+    return None
+
+
 def coerce_resource_config(config: Any) -> Any:
     """Best-effort dict → x402 ``ResourceConfig`` coercion.
 

tests/test_classify_orchestration_error.py

@@ -0,0 +1,112 @@
+"""Tests for ``classify_orchestration_error`` — string-match classification of
+arbitrary thrown errors during the 402 orchestration.
+
+Locked cross-language fixtures shared with the Node sibling at
+``node-commerce/tests/payment/classify_orchestration_error.test.ts``. Both
+files reference identical error messages + expected ClassifiedX402Error
+codes/statuses. Drift in either language (matcher list, case-insensitivity,
+None-on-unknown) fails that language's test against the locked value.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from agentscore_commerce.payment import ClassifiedX402Error, classify_orchestration_error
+
+# Cross-language fixtures: (label, error_message, expected_code_or_None).
+# When the helper returns a ClassifiedX402Error, its `code` matches the third
+# tuple element; `None` means the helper returns None (caller rethrows).
+_FIXTURES: list[tuple[str, str, str | None]] = [
+    # payment_proof_invalid family
+    ("x402version_lowercase", "Unsupported x402Version 3", "payment_proof_invalid"),
+    ("x402version_uppercase", "UNSUPPORTED X402VERSION 3", "payment_proof_invalid"),
+    ("invalid_payment", "Invalid payment payload", "payment_proof_invalid"),
+    ("unsupported_x402", "Unsupported x402 method", "payment_proof_invalid"),
+    # payment_provider_unavailable family
+    ("stripe_lowercase", "Stripe API returned 502", "payment_provider_unavailable"),
+    ("facilitator_lowercase", "Facilitator unreachable", "payment_provider_unavailable"),
+    ("cdp_lowercase", "CDP JWT expired", "payment_provider_unavailable"),
+    ("stripe_uppercase", "STRIPE timeout", "payment_provider_unavailable"),
+    # Unknown — caller rethrows
+    ("database_error", "duplicate key value violates unique constraint", None),
+    ("network_error", "ECONNREFUSED", None),
+    ("empty_string", "", None),
+    ("generic_unknown", "something went wrong", None),
+]
+
+
+@pytest.mark.parametrize(
+    ("label", "message", "expected_code"),
+    _FIXTURES,
+    ids=[label for label, _, _ in _FIXTURES],
+)
+def test_locked_cross_language_fixture(label, message, expected_code) -> None:
+    del label
+    result = classify_orchestration_error(message)
+    if expected_code is None:
+        assert result is None
+    else:
+        assert result is not None
+        assert result.code == expected_code
+
+
+def test_accepts_exception_instance() -> None:
+    """An Exception is stringified via ``str()`` before classification."""
+    err = ValueError("Unsupported x402Version 3")
+    result = classify_orchestration_error(err)
+    assert result is not None
+    assert result.code == "payment_proof_invalid"
+
+
+def test_accepts_baseexception() -> None:
+    """``BaseException`` (e.g. KeyboardInterrupt) is also accepted, for completeness."""
+    err = BaseException("stripe API error")
+    result = classify_orchestration_error(err)
+    assert result is not None
+    assert result.code == "payment_provider_unavailable"
+
+
+def test_returns_400_for_payment_proof_invalid() -> None:
+    result = classify_orchestration_error("x402Version mismatch")
+    assert result is not None
+    assert result.status == 400
+
+
+def test_returns_503_for_payment_provider_unavailable() -> None:
+    result = classify_orchestration_error("Stripe error")
+    assert result is not None
+    assert result.status == 503
+
+
+def test_classified_carries_next_steps() -> None:
+    result = classify_orchestration_error("invalid payment")
+    assert result is not None
+    assert result.next_steps.get("action") == "regenerate_payment_credential"
+    assert "user_message" in result.next_steps
+
+
+def test_provider_classified_carries_retry_after_seconds() -> None:
+    result = classify_orchestration_error("CDP facilitator timeout")
+    assert result is not None
+    assert result.next_steps.get("retry_after_seconds") == 10
+
+
+def test_payment_proof_takes_precedence_when_both_keywords_present() -> None:
+    """An error message containing both pattern families resolves to the first matched."""
+    # "x402Version" is checked first; "stripe" present too but doesn't reach the second branch
+    result = classify_orchestration_error("Unsupported x402Version returned by stripe")
+    assert result is not None
+    assert result.code == "payment_proof_invalid"
+
+
+def test_returns_classified_x402_error_type() -> None:
+    """The return type is the same ``ClassifiedX402Error`` that ``classify_x402_settle_result`` returns."""
+    result = classify_orchestration_error("invalid payment")
+    assert isinstance(result, ClassifiedX402Error)
+
+
+def test_returns_none_for_non_string_non_exception_input() -> None:
+    """Defensive: non-str / non-Exception input returns None."""
+    assert classify_orchestration_error(None) is None  # type: ignore[arg-type]
+    assert classify_orchestration_error(42) is None  # type: ignore[arg-type]