feat(edifabric): surface retry_after_seconds on quota-blocked responses

EdifabricError now carries an optional retry_after_seconds kwarg
parsed from the upstream Retry-After header (API Management quota
policy always sends this on 429/403 quota responses). The spot-check
driver threads the value into each transport_error entry of
spot-check.json so the operator sees the exact moment quota
replenishes rather than parsing it out of the response body.

Adds a MockTransport test confirming retry_after_seconds is set
when the upstream sends Retry-After. Existing callers are
unaffected (retry_after_seconds defaults to None and is keyword-only).

backend/src/cyclone/edifabric.py

@@ -63,13 +63,22 @@ class EdifabricError(RuntimeError):
             ``message`` keys for 4xx, or a string for 5xx / network
             errors. Preserved verbatim so the caller can surface it
             to the operator.
+        retry_after_seconds: When the upstream sent a ``Retry-After``
+            header (the API Management quota policy does), the value
+            in seconds. ``None`` if absent. Quota-blocked callers can
+            surface this so the operator knows when to retry.
     """
 
-    def __init__(self, status_code: int, body: Any) -> None:
+    def __init__(
+        self, status_code: int, body: Any, *, retry_after_seconds: int | None = None
+    ) -> None:
         self.status_code = status_code
         self.body = body
+        self.retry_after_seconds = retry_after_seconds
         body_repr = repr(body) if not isinstance(body, str) else body
         msg = f"Edifabric API returned {status_code}: {body_repr}"
+        if retry_after_seconds is not None:
+            msg += f" (retry after {retry_after_seconds}s)"
         super().__init__(msg)
 
 
@@ -146,7 +155,11 @@ def read_interchange(edi_bytes: bytes, *, api_key: str | None = None) -> dict:
             headers=headers,
         )
     if not (200 <= resp.status_code < 300):
-        raise EdifabricError(resp.status_code, _safe_body(resp))
+        raise EdifabricError(
+            resp.status_code,
+            _safe_body(resp),
+            retry_after_seconds=_retry_after_seconds(resp),
+        )
     data = resp.json()
     if not isinstance(data, list) or not data:
         raise EdifabricError(
@@ -187,7 +200,11 @@ def validate_interchange(x12_json: dict, *, api_key: str | None = None) -> dict:
             headers=headers,
         )
     if not (200 <= resp.status_code < 300):
-        raise EdifabricError(resp.status_code, _safe_body(resp))
+        raise EdifabricError(
+            resp.status_code,
+            _safe_body(resp),
+            retry_after_seconds=_retry_after_seconds(resp),
+        )
     return resp.json()
 
 
@@ -214,3 +231,21 @@ def _safe_body(resp: httpx.Response) -> Any:
     except Exception:  # noqa: BLE001
         text = resp.text
         return text if text else f"<empty {resp.status_code} response>"
+
+
+def _retry_after_seconds(resp: httpx.Response) -> int | None:
+    """Parse the ``Retry-After`` header into integer seconds.
+
+    Returns ``None`` if the header is absent (and the caller should
+    fall back to its own backoff). The upstream API Management sends
+    this header on quota-blocked responses; treating it as authoritative
+    lets the caller sleep exactly until quota replenishes rather than
+    guessing.
+    """
+    raw = resp.headers.get("Retry-After") or resp.headers.get("retry-after")
+    if not raw:
+        return None
+    try:
+        return int(raw)
+    except (TypeError, ValueError):
+        return None

backend/src/cyclone/rebill/spot_check_validate.py

@@ -80,6 +80,12 @@ def validate_spot_check_files(paths: list[Path]) -> list[dict]:
         try:
             result = validate_edi(body)
         except EdifabricError as exc:
+            tx: dict = {
+                "status_code": exc.status_code,
+                "body": exc.body,
+            }
+            if exc.retry_after_seconds is not None:
+                tx["retry_after_seconds"] = exc.retry_after_seconds
             out.append({
                 "file": str(p),
                 "passed": False,
@@ -87,10 +93,7 @@ def validate_spot_check_files(paths: list[Path]) -> list[dict]:
                     f"EdifabricError {exc.status_code}: {exc.body!r}"
                 ),
                 "result": None,
-                "transport_error": {
+                "transport_error": tx,
-                    "status_code": exc.status_code,
-                    "body": exc.body,
-                },
             })
             # Re-raise so the caller can't silently absorb the failure
             # (the original scratch driver caught + fallback'd which

backend/tests/test_edifabric.py

@@ -136,6 +136,28 @@ def test_validate_interchange_raises_on_5xx():
     assert "upstream overloaded" in str(exc_info.value.body)
 
 
+def test_read_interchange_raises_with_retry_after_when_quota_blocked():
+    """On HTTP 403 with a ``Retry-After`` header (API Management quota
+    policy), the raised EdifabricError exposes ``retry_after_seconds``
+    so callers can sleep exactly until quota replenishes."""
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            403,
+            headers={"Retry-After": "50492"},
+            json={"statusCode": 403,
+                  "message": "Out of call volume quota."},
+        )
+
+    _install_factory(handler)
+    with pytest.raises(edifabric.EdifabricError) as exc_info:
+        edifabric.read_interchange(b"ISA*...", api_key=_TEST_KEY)
+    err = exc_info.value
+    assert err.status_code == 403
+    assert err.retry_after_seconds == 50492
+    assert "Out of call volume quota" in str(err.body)
+
+
 # --- validate_edi (composed) ------------------------------------------