"""
Base Boarding API client — Global Payments merchant underwriting REST API.

PRD-HWONB-001 §7.1, PRD-HWONB-002 (token mint + Create Application).

Every public method returns the envelope::

    {
        "ok": bool,
        "data": dict | bytes | None,
        "tsys_errors": list[{"field": str, "message": str}] | None,
        "http_status": int,
    }

No method ever raises for a GP 4xx/5xx response — callers (services.py,
router.py) inspect `ok`/`tsys_errors` and turn them into field-level errors
for the wizard. Only genuinely unexpected local bugs (bad config, programmer
error) raise.

CERT-confirmed corrections baked in here (PRD-HWONB-001 §13.1):
  - N-1: the token endpoint (`POST /accesstoken`) DOES require the
    `X-Gp-Version` header — an earlier PRD draft said it didn't; that was wrong.
  - C-26: the secret hash order is `SHA-512(nonce + client_secret)`, NOT
    reversed.
  - C-27: never send `restricted_token` — the token mints fine without it.

GP error bodies use `errorDetails: [{errorCode, detailedErrorCode,
detailedErrorDescription}, ...]`. `errorCode` is NOT always numeric
(confirmed real example: `"SYSTEM_ERROR_DOWNSTREAM"`) — it is always
treated here as an opaque string, never int-cast.
"""

from __future__ import annotations

import copy
import datetime as _dt
import hashlib
import json
import logging
import re
import time
import uuid
from typing import Any, Dict, List, Optional, Tuple, TypedDict

import httpx

from src.apps.merchant_onboarding.helpers import field_mapping as fm
from src.core.config import settings

logger = logging.getLogger(__name__)

GP_VERSION_HEADER_VALUE = "2024-03-07"


def _api_name_from_path(path: str) -> str:
    """Best-effort human label for a Base Boarding path — the last non-id path
    segment (e.g. `/applications/123/business` -> "business")."""
    segments = [s for s in (path or "").split("?")[0].strip("/").split("/") if s and not s.isdigit()]
    return segments[-1] if segments else (path or "")

# Redis key templates — plain redis-py client, matching the ad-hoc pattern
# used elsewhere in this codebase (role_permissions/dependencies.py,
# feature_control/services.py): `redis.from_url(CELERY_RESULT_BACKEND or
# REDIS_URL, decode_responses=True)`. No dedicated Redis wrapper module
# exists in this codebase to import from.
_TOKEN_CACHE_KEY_TMPL = "onboarding:tsys_token:{env}"
_TOKEN_LOCK_KEY_TMPL = "onboarding:tsys_token:{env}:lock"
_TOKEN_LOCK_TTL_SECONDS = 10
_TOKEN_TTL_SAFETY_MARGIN_SECONDS = 300

# PRD-HWONB-011 §6/C-01 — the resolved rate card (baseRates/perItemFees/
# authorizationFees/pinDebitFees/miscellaneousFees/ensureBill) is
# partner+association-negotiated data that changes rarely, so it's cached per
# (partner, association, processingType, pricingPlanId, optionId) rather than
# re-derived (and re-fetched from 3+ GP lookups) on every push_fees() call.
# PRD-HWONB-014 — association is now merchant-selectable (one of 5 pricing
# tiers), so it MUST be part of the cache key: two merchants on different
# tiers hitting the same (processingType, pricingPlanId, optionId) must not
# be served each other's rate card.
_RATE_CARD_CACHE_KEY_TMPL = (
    "onboarding:fees:rate_card:{env}:{partner}:{association}:{processing_type}:{pricing_plan_id}:{option_id}"
)
_RATE_CARD_CACHE_TTL_SECONDS = 24 * 60 * 60  # 1 day

# An association is "not viable" when every processingType/plan/option triple
# in its feeProcessingDetail catalog is TransFreedom-classified (a live CERT
# trial, 2026-07-20, found association 086386 renamed "Cash Advance" but
# still 100% TransFreedom underneath) — cached the same way/reason as the
# rate card above, keyed only by (partner, association) since it doesn't
# depend on a specific plan/option choice.
_ASSOCIATION_VIABILITY_CACHE_KEY_TMPL = "onboarding:fees:association_viable:{env}:{partner}:{association}"
_ASSOCIATION_VIABILITY_CACHE_TTL_SECONDS = 24 * 60 * 60  # 1 day


class TsysError(TypedDict):
    field: str
    message: str


class MiscFeeRangeError(Exception):
    """Raised by `_resolve_misc_fees` when a merchant-entered named fee override
    falls outside the GP catalog's feeMin/feeMax for that feeCode (p.317/341 of
    the Base Boarding API Guide) — caught in `push_fees` to short-circuit before
    any GP network call, same as the existing rate-card-resolution RuntimeError."""

    def __init__(self, violations: List["TsysError"]) -> None:
        super().__init__("miscellaneousFees override(s) outside catalog range")
        self.violations = violations


class BoardingEnvelope(TypedDict):
    ok: bool
    data: Optional[Any]
    tsys_errors: Optional[List[TsysError]]
    http_status: int


def _ok(data: Any = None, http_status: int = 200) -> BoardingEnvelope:
    return {"ok": True, "data": data, "tsys_errors": None, "http_status": http_status}


def _fail(
    tsys_errors: Optional[List[TsysError]] = None,
    http_status: int = 0,
    data: Any = None,
) -> BoardingEnvelope:
    return {"ok": False, "data": data, "tsys_errors": tsys_errors, "http_status": http_status}


# CERT trial (2026-07-16): GP's `detailedErrorCode` is a generic error-class
# string (e.g. "INVALID_REQUEST_DATA") repeated identically across every
# entry in a single response — it is NOT a per-field identifier, despite
# `_parse_error_body` previously using it as `field`. The real field name
# only appears as the leading words of `detailedErrorDescription`. Each
# pattern below is anchored to a distinct phrasing actually observed in the
# GP developer guide or a live response, rather than one loose universal
# heuristic, to avoid mis-extracting a field name from an unrelated message
# shape we haven't seen. Every pattern's first group is a candidate field
# identifier.
_FIELD_HINT_PATTERNS = [
    re.compile(r"^([A-Za-z][\w.]*)\s+must\b"),  # "Model must match an accepted value: [...]"
    re.compile(r"^([A-Za-z][\w.]*)\s+value\s+is\s+invalid\b", re.IGNORECASE),  # "ebtService value is invalid..."
    re.compile(r"length of\s+([A-Za-z][\w.]*)\s+should\b", re.IGNORECASE),  # "The length of ebtFnsFcsNumber should be 7"
    re.compile(r"\bparameter\s+([A-Za-z][\w.]*)\s*$"),  # "...the mandatory parameter setupFee"
]

# GP's Products/Equipment domain error text uses its own PascalCase field
# names that don't match HubWallet's camelCase JSON body paths 1:1 — translate
# the ones actually observed in a real CERT/live response. An identifier not
# in this table is passed through with only its first letter lowercased
# (GP already uses HubWallet's own camelCase field names verbatim for most
# other sections, e.g. "ebtService") rather than dropped or guessed further.
_GP_ERROR_FIELD_ALIASES = {
    "Model": "model",
    "ConnectionMethod": "connectionMethod",
    "PrintSize": "configurations.printSize",
    "EmvBrand": "pinpadEmvReader.brand",
    "EmvModel": "pinpadEmvReader.model",
    "EmvReaderSource": "pinpadEmvReader.source",
}


def _extract_field_hint(message: str) -> Optional[str]:
    """
    Best-effort extraction of the real field identifier from a GP error
    message, for use as `TsysError.field` instead of the generic
    `detailedErrorCode`. Returns None (never guesses) when no known pattern
    matches — callers must keep falling back to `detailedErrorCode` so this
    is purely additive and never regresses existing behavior.
    """
    stripped = message.strip()
    for pattern in _FIELD_HINT_PATTERNS:
        # `.search()` uniformly — the `^`/`$` anchors already embedded in
        # each pattern above constrain position where it matters.
        match = pattern.search(stripped)
        if match:
            raw = match.group(1)
            if raw in _GP_ERROR_FIELD_ALIASES:
                return _GP_ERROR_FIELD_ALIASES[raw]
            if "." in raw:
                return raw  # already a dotted schema path, e.g. "configurations.batchCloseTime"
            return raw[0].lower() + raw[1:] if raw[:1].isupper() else raw
    return None


def _get_redis_client():
    """
    Best-effort Redis client, mirroring the exact pattern used elsewhere in
    this codebase (role_permissions/dependencies.py, feature_control/services.py).
    Returns None if no Redis URL is configured or the import/connection fails —
    callers must treat a None client as "no cache available", not a hard error.
    """
    try:
        import redis as redis_lib

        url = getattr(settings, "CELERY_RESULT_BACKEND", None) or getattr(
            settings, "REDIS_URL", None
        )
        if url and "redis" in url:
            return redis_lib.from_url(url, decode_responses=True)
    except Exception as exc:
        logger.warning("base_boarding_client: redis unavailable — %s", exc)
    return None


class BaseBoardingClient:
    """
    Thin async HTTP client for Global Payments' Base Boarding API.

    One instance is stateless and safe to reuse/re-instantiate per request —
    all cross-request state (the bearer token) lives in Redis, keyed by
    environment (`TSYS_BOARDING_ENV`), not per-application or per-merchant.
    """

    def __init__(self) -> None:
        self._base_url = (settings.TSYS_BOARDING_BASE_URL or "").rstrip("/")
        self._env = settings.TSYS_BOARDING_ENV or "cert"

    # ------------------------------------------------------------------
    # Token mint / cache
    # ------------------------------------------------------------------

    def _token_cache_key(self) -> str:
        return _TOKEN_CACHE_KEY_TMPL.format(env=self._env)

    def _token_lock_key(self) -> str:
        return _TOKEN_LOCK_KEY_TMPL.format(env=self._env)

    def _cached_token(self) -> Optional[str]:
        r = _get_redis_client()
        if not r:
            return None
        try:
            return r.get(self._token_cache_key())
        except Exception as exc:
            logger.warning("base_boarding_client: token cache read failed — %s", exc)
            return None

    def _invalidate_cached_token(self) -> None:
        r = _get_redis_client()
        if not r:
            return
        try:
            r.delete(self._token_cache_key())
        except Exception as exc:
            logger.warning("base_boarding_client: token cache invalidate failed — %s", exc)

    def _cache_token(self, token: str, ttl_seconds: int) -> None:
        r = _get_redis_client()
        if not r:
            return
        try:
            r.setex(self._token_cache_key(), ttl_seconds, token)
        except Exception as exc:
            logger.warning("base_boarding_client: token cache write failed — %s", exc)

    async def _get_token(self, merchant_id: Optional[int] = None) -> Optional[str]:
        """Return a valid cached token, minting a fresh one (single-flight) if absent."""
        cached = self._cached_token()
        if cached:
            return cached

        r = _get_redis_client()
        lock_key = self._token_lock_key()
        got_lock = True
        if r is not None:
            try:
                got_lock = bool(r.set(lock_key, "1", nx=True, ex=_TOKEN_LOCK_TTL_SECONDS))
            except Exception as exc:
                logger.warning("base_boarding_client: token lock acquire failed — %s", exc)
                got_lock = True  # degrade to "no locking" rather than blocking mint entirely

        if not got_lock:
            # Another process is minting right now — brief wait then re-check cache once.
            import asyncio

            await asyncio.sleep(0.3)
            cached = self._cached_token()
            if cached:
                return cached
            # Still nothing cached — fall through and mint anyway rather than hang;
            # worst case is a harmless duplicate mint.

        try:
            envelope = await self.mint_token(merchant_id=merchant_id)
            if not envelope["ok"] or not envelope["data"]:
                return None
            return envelope["data"].get("token")
        finally:
            if r is not None and got_lock:
                try:
                    r.delete(lock_key)
                except Exception:
                    pass

    async def mint_token(self, merchant_id: Optional[int] = None) -> BoardingEnvelope:
        """
        POST /accesstoken — mint a fresh Base Boarding bearer token and cache it.

        PRD-HWONB-002 §2. Secret = SHA-512(nonce + client_secret) hex digest —
        CERT-confirmed order (C-26); reversing it returns 403 ACTION_NOT_AUTHORIZED.
        `restricted_token` is never sent (C-27 — mints fine without it).
        `X-Gp-Version` IS required on this call (N-1 correction).

        Caches the returned token in Redis under `onboarding:tsys_token:{env}`,
        TTL = the live `seconds_to_expire` response value minus a 300s safety
        margin — never a hardcoded assumption tied to `interval_to_expire`
        (a documented GP quirk pairs `DAY` with `seconds_to_expire: 604799`
        in at least one recorded example).
        """
        nonce = uuid.uuid4().hex
        secret = hashlib.sha512((nonce + settings.TSYS_BOARDING_CLIENT_SECRET).encode()).hexdigest()

        body = {
            "app_id": settings.TSYS_BOARDING_APP_ID,
            "secret": secret,
            "grant_type": "client_credentials",
            "nonce": nonce,
            "interval_to_expire": "DAY",
        }
        headers = {
            "Content-Type": "application/json",
            "X-Gp-Version": GP_VERSION_HEADER_VALUE,
        }

        url = f"{self._base_url}/accesstoken"
        _t0 = time.perf_counter()
        _ms = lambda: int((time.perf_counter() - _t0) * 1000)  # noqa: E731
        try:
            async with httpx.AsyncClient(timeout=30.0) as client:
                resp = await client.post(url, json=body, headers=headers)
        except httpx.HTTPError as exc:
            logger.error("base_boarding_client.mint_token: network error — %s", exc)
            self._log_call(
                merchant_id=merchant_id, application_id=None, method="POST", path="/accesstoken",
                url=url, headers=headers, body=body, response=None, error=exc,
                duration_ms=_ms(), api_name="Authentication",
            )
            return _fail(tsys_errors=[{"field": "_system", "message": "Unable to reach Global Payments."}])

        if resp.status_code >= 500:
            logger.error(
                "base_boarding_client.mint_token: GP 5xx status=%s body=%s",
                resp.status_code,
                resp.text[:2000],
            )
            self._log_call(
                merchant_id=merchant_id, application_id=None, method="POST", path="/accesstoken",
                url=url, headers=headers, body=body, response=resp,
                duration_ms=_ms(), api_name="Authentication",
            )
            return _fail(
                tsys_errors=[{"field": "_system", "message": "Global Payments is temporarily unavailable."}],
                http_status=resp.status_code,
            )

        if resp.status_code >= 400:
            tsys_errors = self._parse_error_body(resp)
            logger.error(
                "base_boarding_client.mint_token: GP 4xx status=%s errors=%s",
                resp.status_code,
                tsys_errors,
            )
            self._log_call(
                merchant_id=merchant_id, application_id=None, method="POST", path="/accesstoken",
                url=url, headers=headers, body=body, response=resp,
                duration_ms=_ms(), api_name="Authentication",
            )
            return _fail(tsys_errors=tsys_errors, http_status=resp.status_code)

        self._log_call(
            merchant_id=merchant_id, application_id=None, method="POST", path="/accesstoken",
            url=url, headers=headers, body=body, response=resp,
            duration_ms=_ms(), api_name="Authentication",
        )
        payload = resp.json()
        token = payload.get("token")
        seconds_to_expire = payload.get("seconds_to_expire")
        if token and isinstance(seconds_to_expire, (int, float)) and seconds_to_expire > _TOKEN_TTL_SAFETY_MARGIN_SECONDS:
            ttl = int(seconds_to_expire) - _TOKEN_TTL_SAFETY_MARGIN_SECONDS
            self._cache_token(token, ttl)
        elif token:
            # Defensive fallback only — the live response should always carry
            # seconds_to_expire. TSYS_BOARDING_TOKEN_TTL_SECONDS exists purely
            # as a safety-net TTL for this unexpected-response case, never as
            # the primary TTL source.
            fallback_ttl = max(
                settings.TSYS_BOARDING_TOKEN_TTL_SECONDS - _TOKEN_TTL_SAFETY_MARGIN_SECONDS, 60
            )
            logger.warning(
                "base_boarding_client.mint_token: seconds_to_expire missing/invalid in response — "
                "using fallback TTL=%ss",
                fallback_ttl,
            )
            self._cache_token(token, fallback_ttl)

        return _ok(data=payload, http_status=resp.status_code)

    # ------------------------------------------------------------------
    # Generic request wrapper
    # ------------------------------------------------------------------

    def _parse_error_body(self, resp: httpx.Response) -> List[TsysError]:
        """
        Parse GP's `errorDetails: [{errorCode, detailedErrorCode,
        detailedErrorDescription}]` shape into `[{field, message}]`.

        `errorCode`/`detailedErrorCode` are always treated as opaque strings —
        never int-cast (confirmed non-numeric values like
        "SYSTEM_ERROR_DOWNSTREAM" exist in real GP responses).

        CERT trial (2026-07-16): `detailedErrorCode` is a generic error-class
        string (e.g. "INVALID_REQUEST_DATA", repeated identically across
        every entry in a single response) — NOT a per-field name, contrary
        to this method's prior behavior of using it as `field`. The real
        field identifier only exists inside the free-text
        `detailedErrorDescription`, so `field` is now `_extract_field_hint()`'s
        best-effort extraction from that text when confidently matched,
        falling back to the detailedErrorCode (or "_system") exactly as
        before when it isn't.
        """
        try:
            body = resp.json()
        except Exception:
            snippet = (resp.text or "").strip()[:500]
            msg = (
                f"Global Payments returned an unreadable error ({resp.status_code}): {snippet}"
                if snippet
                else f"Global Payments returned an unreadable error ({resp.status_code})."
            )
            return [{"field": "_system", "message": msg}]

        details = body.get("errorDetails") or []
        if not details:
            # GP rejected without the standard errorDetails array — don't discard
            # the reason (it's the only thing that tells the merchant what to fix).
            # Pull it from common alternate body shapes, else the raw text.
            reason = (
                body.get("message")
                or body.get("error")
                or body.get("responseText")
                or (resp.text or "").strip()[:500]
            )
            msg = (
                f"Global Payments rejected the request: {reason}"
                if reason
                else "Global Payments rejected the request."
            )
            return [{"field": "_system", "message": str(msg)}]

        errors: List[TsysError] = []
        for d in details:
            code = str(d.get("detailedErrorCode") or d.get("errorCode") or "_system")
            message = str(d.get("detailedErrorDescription") or d.get("errorCode") or "Unknown error")
            field = _extract_field_hint(message) or code
            errors.append({"field": field, "message": message})
        return errors


    def _log_call(
        self,
        *,
        merchant_id: Optional[int],
        application_id: Optional[int],
        method: str,
        path: str,
        url: str,
        headers: Dict[str, Any],
        body: Any,
        response: Optional[httpx.Response] = None,
        error: Optional[Exception] = None,
        duration_ms: Optional[int] = None,
        api_name: Optional[str] = None,
    ) -> None:
        """
        Persist one masked Base Boarding request/response pair to
        `onboarding_api_call_logs`. Unwraps this client's httpx call kwargs
        (`json`/`files`/`params`) into a masked request body, then delegates
        the mask-and-persist to the shared `record_api_call_log` writer
        (service="Base CRM").
        """
        request_body: Any = None
        if isinstance(body, dict):
            if "json" in body:
                request_body = body.get("json")
            elif "files" in body:
                summary: Dict[str, Any] = {}
                # files may be a dict (single upload_attachment) or a list of
                # (field, value) tuples (upload_attachments — the `documents`
                # key repeats for multi-file multipart, so a dict can't hold it).
                _files = body.get("files") or {}
                _items = _files.items() if isinstance(_files, dict) else _files
                for field, value in _items:
                    if isinstance(value, tuple) and len(value) == 3:
                        fname, fbytes, ctype = value
                        summary[field] = {
                            "filename": fname,
                            "content_type": ctype,
                            "size_bytes": len(fbytes) if hasattr(fbytes, "__len__") else None,
                        }
                    else:
                        summary[field] = str(value)
                if body.get("params"):
                    summary["_query_params"] = body["params"]
                request_body = summary
            elif "params" in body:
                request_body = body.get("params")
            elif body:
                request_body = body

        from src.apps.merchant_onboarding.helpers.log_writer import record_api_call_log

        record_api_call_log(
            service="Base CRM",
            api_name=api_name or _api_name_from_path(path),
            merchant_id=merchant_id,
            application_id=application_id,
            method=method,
            path=path,
            url=url,
            request_headers=headers,
            request_body=request_body,
            response=response,
            error=error,
            duration_ms=duration_ms,
        )

    async def _request(
        self,
        method: str,
        path: str,
        *,
        merchant_id: Optional[int] = None,
        application_id: Optional[int] = None,
        _retried_after_401: bool = False,
        **kwargs: Any,
    ) -> BoardingEnvelope:
        """
        Private helper wrapping httpx.AsyncClient for every authenticated
        Base Boarding call (everything except `mint_token`).

        Injects `Authorization: Bearer <token>` + `X-Gp-Version: 2024-03-07`
        on every call. On a 401: invalidates the cached token, re-mints once,
        and replays the request once. A second 401 becomes a `tsys_errors`
        system error — no further retry.
        """
        token = await self._get_token(merchant_id=merchant_id)
        if not token:
            self._log_call(
                merchant_id=merchant_id, application_id=application_id, method=method, path=path,
                url=f"{self._base_url}{path}", headers={}, body=kwargs, response=None,
                error=Exception("Could not obtain a Global Payments access token."),
            )
            return _fail(tsys_errors=[{"field": "_system", "message": "Could not obtain a Global Payments access token."}])

        headers = kwargs.pop("headers", {}) or {}
        headers = {
            **headers,
            "Authorization": f"Bearer {token}",
            "X-Gp-Version": GP_VERSION_HEADER_VALUE,
        }

        url = f"{self._base_url}{path}"
        _t0 = time.perf_counter()
        _ms = lambda: int((time.perf_counter() - _t0) * 1000)  # noqa: E731
        try:
            async with httpx.AsyncClient(timeout=30.0) as client:
                resp = await client.request(method, url, headers=headers, **kwargs)
        except httpx.HTTPError as exc:
            logger.error("base_boarding_client._request: network error %s %s — %s", method, path, exc)
            self._log_call(
                merchant_id=merchant_id, application_id=application_id, method=method, path=path,
                url=url, headers=headers, body=kwargs, response=None, error=exc, duration_ms=_ms(),
            )
            return _fail(tsys_errors=[{"field": "_system", "message": "Unable to reach Global Payments."}])

        if resp.status_code == 401:
            self._log_call(
                merchant_id=merchant_id, application_id=application_id, method=method, path=path,
                url=url, headers=headers, body=kwargs, response=resp, duration_ms=_ms(),
            )
            if _retried_after_401:
                logger.error(
                    "base_boarding_client._request: second consecutive 401 on %s %s — giving up",
                    method,
                    path,
                )
                return _fail(
                    tsys_errors=[{"field": "_system", "message": "Global Payments authentication failed."}],
                    http_status=401,
                )
            logger.warning(
                "base_boarding_client._request: 401 on %s %s — invalidating token and retrying once",
                method,
                path,
            )
            self._invalidate_cached_token()
            return await self._request(
                method, path, merchant_id=merchant_id, application_id=application_id,
                _retried_after_401=True, **kwargs,
            )

        if resp.status_code >= 500:
            logger.error(
                "base_boarding_client._request: GP 5xx %s %s status=%s body=%s",
                method,
                path,
                resp.status_code,
                resp.text[:2000],
            )
            self._log_call(
                merchant_id=merchant_id, application_id=application_id, method=method, path=path,
                url=url, headers=headers, body=kwargs, response=resp, duration_ms=_ms(),
            )
            return _fail(
                tsys_errors=[{"field": "_system", "message": "Global Payments is temporarily unavailable."}],
                http_status=resp.status_code,
            )

        if resp.status_code >= 400:
            tsys_errors = self._parse_error_body(resp)
            logger.info(
                "base_boarding_client._request: GP 4xx %s %s status=%s errors=%s",
                method,
                path,
                resp.status_code,
                tsys_errors,
            )
            self._log_call(
                merchant_id=merchant_id, application_id=application_id, method=method, path=path,
                url=url, headers=headers, body=kwargs, response=resp, duration_ms=_ms(),
            )
            return _fail(tsys_errors=tsys_errors, http_status=resp.status_code)

        content_type = resp.headers.get("content-type", "")
        if "application/json" in content_type:
            try:
                data = resp.json()
            except Exception:
                data = None
        else:
            data = resp.content

        self._log_call(
            merchant_id=merchant_id, application_id=application_id, method=method, path=path,
            url=url, headers=headers, body=kwargs, response=resp, duration_ms=_ms(),
        )
        return _ok(data=data, http_status=resp.status_code)

    # ------------------------------------------------------------------
    # PRD-HWONB-002 — Application Bootstrap
    # ------------------------------------------------------------------

    async def create_application(
        self, *, association: str, lead_source: str, sales_rep: int, merchant_id: Optional[int] = None, application_id: Optional[int] = None
    ) -> BoardingEnvelope:
        """
        POST /applications — creates the GP underwriting application shell.

        PRD-HWONB-015: `association`/`lead_source`/`sales_rep` are the
        merchant's own choices from the Application Setup step (options
        sourced from GP's `masterRecord` lookups) — the caller
        (`services._save_application_setup_section`) is the only one that
        calls this, always with real merchant-submitted values, never a
        platform-wide default (this supersedes PRD-HWONB-002's fully
        config-driven body and PRD-HWONB-014's "association is a bootstrap
        default, patched in later" flow). `partner` (fixed "HUBWALLET"),
        `merchantApplicationType` (fixed "NEW_MERCHANT"), and `sBank` (fixed
        sponsor bank) remain config-driven — never merchant-facing.
        Returns envelope with `data["appId"]` on success (int).
        """
        body: Dict[str, Any] = {
            "partner": settings.TSYS_BOARDING_PARTNER,
            "association": association,
            "leadSource": lead_source,
            # GP's Create Application contract types salesRep as a JSON
            # Integer (Developer Guide V1.9 p.86: `"salesRep": 46`,
            # unquoted) — the validated request schema already guarantees
            # this is an int, so no cast is needed here.
            "salesRep": sales_rep,
            "merchantApplicationType": "NEW_MERCHANT",
        }
        if settings.TSYS_BOARDING_PROSPECT_ID:
            body["prospectId"] = settings.TSYS_BOARDING_PROSPECT_ID
        if settings.TSYS_BOARDING_SPONSOR_BANK:
            body["sBank"] = settings.TSYS_BOARDING_SPONSOR_BANK

        return await self._request("POST", "/applications", json=body, merchant_id=merchant_id, application_id=application_id)

    async def patch_configuration(
        self, app_id: int, association: str, *, lead_source: str, sales_rep: int, merchant_id: Optional[int] = None, application_id: Optional[int] = None
    ) -> BoardingEnvelope:
        """
        PATCH /applications/{appId}/configuration.

        GP's own contract replaces the whole configuration sub-object except
        `partner` (Developer Guide p.90). Dormant as of PRD-HWONB-015 — the
        Pricing Tier step that called this (PRD-HWONB-014) has been retired
        from the wizard now that `association` (along with `leadSource`/
        `salesRep`) is captured upfront at Create Application time, and GP
        has no other section corresponding to a later change of these three
        values. Kept for a possible future support/admin "change
        association" action — such a caller would need to pass the
        application's own `tsys_lead_source`/`tsys_sales_rep`, not a
        platform-wide settings default (those no longer exist as settings).
        """
        body: Dict[str, Any] = {
            "partner": settings.TSYS_BOARDING_PARTNER,
            "association": association,
            "leadSource": lead_source,
            "salesRep": sales_rep,
        }
        if settings.TSYS_BOARDING_PROSPECT_ID:
            body["prospectId"] = settings.TSYS_BOARDING_PROSPECT_ID

        return await self._request("PATCH", f"/applications/{app_id}/configuration", json=body, merchant_id=merchant_id, application_id=application_id)

    async def get_master_record(
        self, record_type: str, partner: str, association: Optional[str] = None, merchant_id: Optional[int] = None, application_id: Optional[int] = None
    ) -> BoardingEnvelope:
        """
        GET /masterRecord/{record_type} — used by lookups for sicCode,
        ownershipType, businessType, contactType, country, and (with the
        extra `association` param) salesRep.

        Response shape (CERT-confirmed, C-12):
        `{<type>: [{<type>: <label>, <type>Value: <code>}, ...]}`.
        """
        params: Dict[str, str] = {"partner": partner}
        if association:
            params["association"] = association
        return await self._request("GET", f"/masterRecord/{record_type}", params=params, merchant_id=merchant_id, application_id=application_id)

    # ------------------------------------------------------------------
    # Stubs — implemented by the assigned vertical-slice developer for each
    # step PRD. Signatures are fixed now so parallel developers can write
    # callers against them; paths below are the best-available reading of
    # the step PRDs and PRD-001 §13.1's CERT findings, and may need small
    # corrections (e.g. exact query params) once that developer verifies
    # against CERT directly.
    # ------------------------------------------------------------------

    async def push_business(self, app_id: int, payload: Dict[str, Any], merchant_id: Optional[int] = None, application_id: Optional[int] = None) -> BoardingEnvelope:
        """
        POST /applications/{appId}/business — PRD-HWONB-003 §3/§3A.

        (The stub's original docstring said PUT; PRD-HWONB-003's own BaseCRM
        Endpoint Reference table — the authoritative source — confirms this
        is a POST: `POST /basecrm/applications/{appId}/business`. Corrected
        here, no other stub signature/behavior touched.)

        `payload` is the already-validated `BusinessInformationRequest.to_gp_payload()`
        dict — this method does no reshaping of its own (PRD-HWONB-003 §3: the
        HubWallet schema mirrors the GP contract near-verbatim for this section).
        """
        return await self._request("POST", f"/applications/{app_id}/business", json=payload, merchant_id=merchant_id, application_id=application_id)

    async def push_processing(self, app_id: int, payload: Dict[str, Any], merchant_id: Optional[int] = None, application_id: Optional[int] = None) -> BoardingEnvelope:
        """
        POST /applications/{appId}/processingInformation — PRD-HWONB-004 §3/§3A.

        (Same POST-not-PUT correction as push_business, per PRD-HWONB-004's
        BaseCRM Endpoint Reference table.)

        `payload` is `ProcessingInformationRequest.to_gp_payload()` — three
        nested objects (`applicationProcessing`, `applicationPatriotAct`,
        `applicationMerchantSurvey`), no reshaping.
        """
        return await self._request(
            "POST", f"/applications/{app_id}/processingInformation", json=payload, merchant_id=merchant_id, application_id=application_id
        )

    async def push_addresses(self, app_id: int, payload: Dict[str, Any], merchant_id: Optional[int] = None, application_id: Optional[int] = None) -> BoardingEnvelope:
        """
        POST /applications/{appId}/addresses — PRD-HWONB-005 (Addresses).

        Keyed-object envelope CERT-confirmed (C-11): `{dba, legal, mailing,
        chargeback}`. `payload` must already be shaped exactly like that by
        the caller (services.py::_build_gp_addresses_payload) — when a
        sub-object's `isSameAsDba` is true, that sub-object must already be
        reduced to `{"isSameAsDba": true}` before it reaches this method;
        this client method does no reshaping of its own, matching every
        other `push_*` method's contract (payload in, envelope out).
        """
        return await self._request("POST", f"/applications/{app_id}/addresses", json=payload, merchant_id=merchant_id, application_id=application_id)

    async def lookup_zipcode(self, partner: str, zip_code: str, merchant_id: Optional[int] = None, application_id: Optional[int] = None) -> BoardingEnvelope:
        """
        GET /masterRecord/zipcode?partner={partner}&zipCode={zip} —
        PRD-HWONB-005, CERT-corrected path/params (N-3): the query param is
        `zipCode` (not `zip`), and `partner` is required — omitting either
        returns GP error `40251`. A bare `/zipcode` path (no `masterRecord`
        prefix) returns `40252 INVALID_ACTION`.
        """
        params = {"partner": partner, "zipCode": zip_code}
        return await self._request("GET", "/masterRecord/zipcode", params=params, merchant_id=merchant_id, application_id=application_id)

    async def push_account(self, app_id: int, payload: Dict[str, Any], merchant_id: Optional[int] = None, application_id: Optional[int] = None) -> BoardingEnvelope:
        """
        POST /applications/{appId}/accounts — PRD-HWONB-006 (Bank Accounts).

        Per-account create, NOT a batch call and NOT a `{accountId}` path
        suffix — GP generates and returns the account identifier on create
        (§4A). Callers (services.py) invoke this once per row in the
        submitted account list, never with the full array in one call.
        """
        return await self._request("POST", f"/applications/{app_id}/accounts", json=payload, merchant_id=merchant_id, application_id=application_id)

    async def get_accounts(self, app_id: int, merchant_id: Optional[int] = None, application_id: Optional[int] = None) -> BoardingEnvelope:
        """
        GET /applications/{appId}/accounts — PRD-HWONB-006 Show/Sync. Used
        on wizard resume and as a manual reconcile action to detect drift
        between HubWallet's local rows and GP's actual stored copy (§4).
        """
        return await self._request("GET", f"/applications/{app_id}/accounts", merchant_id=merchant_id, application_id=application_id)

    async def push_owner(self, app_id: int, payload: Dict[str, Any], merchant_id: Optional[int] = None, application_id: Optional[int] = None) -> BoardingEnvelope:
        """
        POST /applications/{appId}/owners — PRD-HWONB-007 §4/§4A.

        Create is per-owner and takes NO `{ownerId}` in the path — GP
        generates and returns the owner identifier on create (the client
        never supplies one). Called once per owner in the list by
        services.py; `payload` is a single owner's GP-shaped (camelCase)
        dict, not the whole array.
        """
        return await self._request("POST", f"/applications/{app_id}/owners", json=payload, merchant_id=merchant_id, application_id=application_id)

    async def get_owners(self, app_id: int, merchant_id: Optional[int] = None, application_id: Optional[int] = None) -> BoardingEnvelope:
        """GET /applications/{appId}/owners — PRD-HWONB-007 §4/§4A Show/Sync reconcile."""
        return await self._request("GET", f"/applications/{app_id}/owners", merchant_id=merchant_id, application_id=application_id)

    async def push_card_types(self, app_id: int, payload: Dict[str, Any], merchant_id: Optional[int] = None, application_id: Optional[int] = None) -> BoardingEnvelope:
        """
        POST /applications/{appId}/cardTypes — PRD-HWONB-008 §3/§3A.

        Flat object, 1:1 mapping — `payload` is already GP-shaped
        (schemas.card_types.CardTypesRequest.model_dump(by_alias=True,
        exclude_none=True)), no reshaping here.
        """
        return await self._request("POST", f"/applications/{app_id}/cardTypes", json=payload, merchant_id=merchant_id, application_id=application_id)

    async def push_product(self, app_id: int, payload: Dict[str, Any], merchant_id: Optional[int] = None, application_id: Optional[int] = None) -> BoardingEnvelope:
        """
        POST /applications/{appId}/products — PRD-HWONB-009.

        CERT-confirmed (C-16 round 2): there is NO trailing `{productId}` on
        create — an earlier draft's `/products/{id}` path is what caused the
        original 502s. GP generates and returns a `terminalNumber` in the
        response body (`data["terminalNumber"]`) — `services.save_section()`
        persists this into `section_state["products"]` for any future PATCH.
        `payload` must be the full discriminated-union body built from
        `schemas.products.StandaloneProduct` / `IntegratedProduct`
        (`.model_dump()`), including the full 13-boolean `features` block for
        standalone (C-16) — a partial body reproduces the earlier 502.
        """
        return await self._request("POST", f"/applications/{app_id}/products", json=payload, merchant_id=merchant_id, application_id=application_id)

    async def get_standalone_equipment_catalog(
        self,
        partner: str,
        association: Optional[str] = None,
        *,
        brand: Optional[str] = None,
        model: Optional[str] = None,
        industry: Optional[str] = None,
        application: Optional[str] = None,
        source: Optional[str] = None,
        merchant_id: Optional[int] = None, application_id: Optional[int] = None,
    ) -> BoardingEnvelope:
        """
        GET /product/standalone/equipment/list (or the filtered
        /product/standalone/equipment when brand/model/industry/application/
        source are supplied) — PRD-HWONB-009 §2, shape confirmed C-16.

        `association` defaults to `settings.TSYS_BOARDING_ASSOCIATION` when
        omitted so this remains callable with just `partner` (the router's
        generic `/onboarding/lookups/{type}` dispatcher only passes
        `partner` positionally today).

        Response shape (C-16): `{"equipment": [{"brand", "model",
        "sourceBillTo": {...}, "industryApplication": {...}}, ...]}` from
        `/list`. The non-`/list` `standalone/equipment` requires
        brand/model/industry/application/source query params or GP returns
        `40024` — use it only once a brand is selected and a caller wants a
        narrower/validated lookup (e.g. confirming a model+industry
        combination), never for the initial catalog populate.
        """
        association = association or settings.TSYS_BOARDING_ASSOCIATION
        params: Dict[str, str] = {"partner": partner, "association": association}
        filters = {
            "brand": brand,
            "model": model,
            "industry": industry,
            "application": application,
            "source": source,
        }
        filters = {k: v for k, v in filters.items() if v}
        if not filters:
            return await self._request(
                "GET", "/product/standalone/equipment/list", params=params, merchant_id=merchant_id, application_id=application_id
            )

        params.update(filters)
        envelope = await self._request(
            "GET", "/product/standalone/equipment", params=params, merchant_id=merchant_id, application_id=application_id
        )
        if envelope["ok"] and isinstance(envelope["data"], dict):
            equipment = envelope["data"].get("Equipment") or envelope["data"].get("equipment")
            if isinstance(equipment, dict):
                equipment["connectionMethods"] = self._derive_connection_methods(
                    equipment.get("communicationWithPOS") or []
                )
        return envelope

    @staticmethod
    def _derive_connection_methods(labels: List[str]) -> List[Dict[str, Optional[str]]]:
        """
        Maps GP's `communicationWithPOS` descriptive labels (e.g. "IP/SSL",
        "WiFi") to the single-letter connectionMethod codes `Create Product`
        requires, via `field_mapping.CONNECTION_METHOD_LABEL_TO_CODE` — see
        that dict's docstring for which labels are CERT-confirmed vs.
        guide-only. An unrecognized label is kept with `code: None` (logged)
        rather than dropped or guessed — the UI skips code-less entries,
        degrading to "connectionMethod not offered" instead of risking a
        wrong submission.
        """
        derived: List[Dict[str, Optional[str]]] = []
        for label in labels:
            code = fm.CONNECTION_METHOD_LABEL_TO_CODE.get(label)
            if code is None:
                logger.warning(
                    "get_standalone_equipment_catalog: unrecognized communicationWithPOS "
                    "label %r — extend CONNECTION_METHOD_LABEL_TO_CODE once confirmed against "
                    "CERT; omitting a mapped code for this entry.",
                    label,
                )
            derived.append({"label": label, "code": code})
        return derived

    async def get_integrated_product_catalog(
        self, partner: str, association: Optional[str] = None, merchant_id: Optional[int] = None, application_id: Optional[int] = None
    ) -> BoardingEnvelope:
        """
        GET /product/integrated — PRD-HWONB-009 §4, shape confirmed C-16:
        `{"integratedProduct": [{"productName", "integrationProductID",
        "configuration": {"monthlyFee", "setUpFee", "allowedFeature": {...},
        "equipLimit", "equipMin"}}, ...]}`. The catalog's identifier field is
        `integrationProductID` (int, e.g. 106 "WebPASS", 128) — this is the
        value the wizard writes into the product's `integrationProductID`
        field (see schemas.products module docstring for the
        integrationSwId-vs-integrationProductID write-field judgment call —
        unconfirmed either way since only standalone was CERT-posted).
        """
        association = association or settings.TSYS_BOARDING_ASSOCIATION
        return await self._request(
            "GET", "/product/integrated", params={"partner": partner, "association": association}, merchant_id=merchant_id, application_id=application_id
        )

    async def push_training_activation(self, app_id: int, payload: Dict[str, Any], merchant_id: Optional[int] = None, application_id: Optional[int] = None) -> BoardingEnvelope:
        """
        POST /applications/{appId}/trainingActivation — PRD-HWONB-010 §5,
        shape confirmed C-18. Flat object (plus conditional shipping-address
        sub-fields when `equipShippedTo="Other"`) — no reshaping needed, per
        `schemas.training_activation.TrainingActivationRequest.to_gp_payload()`.
        """
        return await self._request("POST", f"/applications/{app_id}/trainingActivation", json=payload, merchant_id=merchant_id, application_id=application_id)

    async def get_training_activation_catalog(
        self, partner: str, association: Optional[str] = None, merchant_id: Optional[int] = None, application_id: Optional[int] = None
    ) -> BoardingEnvelope:
        """
        GET /product/trainingAndActivationDetail — PRD-HWONB-010 §3, shape
        confirmed C-18: `{"trainingAndActivationDetail": {"merchantTrainedBy":
        [...], "equipmentShippedTo": [...], "welcomeKitEmailedTo": [...]}}`.
        Needs `partner` AND `association` (C-18) — `association` defaults to
        `settings.TSYS_BOARDING_ASSOCIATION` when the caller (the router's
        generic single-arg lookup dispatcher) only supplies `partner`.
        """
        association = association or settings.TSYS_BOARDING_ASSOCIATION
        return await self._request(
            "GET",
            "/product/trainingAndActivationDetail",
            params={"partner": partner, "association": association},
            merchant_id=merchant_id, application_id=application_id,
        )

    # ------------------------------------------------------------------
    # PRD-HWONB-011 — Pricing & Fees (implemented; AC-14 hard blocker)
    # ------------------------------------------------------------------

    async def push_fees(
        self,
        app_id: int,
        payload: Dict[str, Any],
        association: Optional[str] = None,
        business_data: Optional[Dict[str, Any]] = None,
        card_types_data: Optional[Dict[str, Any]] = None,
        products_data: Optional[Dict[str, Any]] = None,
        named_misc_fee_overrides: Optional[Dict[str, float]] = None,
        merchant_id: Optional[int] = None, application_id: Optional[int] = None,
    ) -> BoardingEnvelope:
        """
        POST /applications/{appId}/fees — PRD-HWONB-011.

        `association` (PRD-HWONB-014) should be the merchant's own selected
        pricing tier (`application.tsys_association`) — callers should pass
        it explicitly rather than relying on the `settings` fallback below,
        which exists only so a pre-selection call (shouldn't normally
        happen — Fees is gated on Pricing Tier being pushed first) doesn't
        hard-fail.

        `payload` is the merchant-facing GP-shaped fragment already built by
        `schemas.fees.FeesRequest.to_gp_payload()` — processingType/
        pricingPlanId/optionId/amexPricingInd/the mandatory opt-in booleans/
        wireless/ACH/unsupported-POS/Genius selectors. This method is
        responsible for merging in the server-injected rate-card fields
        (§6/C-01) via `_resolve_rate_card()` before POSTing — the merchant
        never supplies baseRates/perItemFees/authorizationFees/pinDebitFees/
        miscellaneousFees/additionalFees.ensureBill directly.

        ================================================================
        READ BEFORE "FIXING" A FAILED 201 HERE
        ================================================================
        CERT (2026-07-14, TEST833/plan-30/option-41 "TransFreedom") returned
        INCONSISTENT validation errors across byte-identical POST bodies —
        ensureBill/pinDebit rules appeared to toggle between pass/fail
        across requests in the same session (PRD-HWONB-001 §13.1 C-01,
        PRD-HWONB-011 §6). Root-caused (Base Boarding API Guide pp.
        313-328, read in full 2026-07-20): `additionalFees.ensureBill` was
        wrongly nested (GP's own Sample Request/Response show
        setUpFee/monthlyFee/deliveryFee as FLAT siblings under
        `additionalFees`, no `ensureBill` wrapper — see `_resolve_ensure_bill`
        and this method's `additionalFees` assembly below), and TransFreedom
        was wrongly treated as unsendable rather than zero-filled (see
        `_apply_trans_freedom_zero_fill`) — not CERT flakiness.
        If a future CERT trial 400s here:
          1. First diff the exact request body (`envelope["sent_payload"]`)
             against the doc's Sample Request (pp. 324-326) field-by-field.
          2. Re-run against a REAL merchant rate sheet / a different test
             partner before assuming `_resolve_rate_card` needs a rewrite —
             its field-by-field rules (rates > 0.00, amex omitted when
             amexPricingInd=false, miscellaneousFees non-empty with a real
             feeCode, pinDebitFees > 0) are exactly the set of checks CERT
             confirmed GP performs (C-01 round 2). Debug/extend this
             method and `_resolve_rate_card`, don't replace them wholesale.
          3. The one genuinely unverified piece is `RATE_CARD_FIELD_NAME_MAP`
             in helpers/field_mapping.py beyond the "AllCardTypes" stem —
             that mapping is inferred, not CERT-confirmed, and is the most
             likely real gap once a plan with more than one rate-card field
             is exercised.
        ================================================================
        """
        processing_type = payload.get("processingType")
        pricing_plan_id = payload.get("pricingPlanId")
        option_id = payload.get("optionId")
        amex_pricing_ind = bool(payload.get("amexPricingInd", False))

        if processing_type is None or pricing_plan_id is None or option_id is None:
            return _fail(
                tsys_errors=[{
                    "field": "_system",
                    "message": "processingType/pricingPlanId/optionId are required to resolve the rate card.",
                }]
            )

        # p.322/347 — "geniusPlanName ... is required when a Genius product is
        # added to the application"; conversely GP rejects it with no Genius
        # product on the application. Enforced here, before any GP call, since
        # the merchant-facing FeesRequest schema has no way to see the
        # Products & Equipment section's saved state.
        if payload.get("geniusPlan") and (products_data or {}).get("productType") != "genius":
            return _fail(
                tsys_errors=[{
                    "field": "addOnPlan",
                    "message": "Add a Genius product in Products & Equipment before selecting a Genius Plan here.",
                }]
            )

        # Base Boarding API Guide pp. 320-321 — dataProtectionPerItemFee
        # "becomes mandatory when tokenization is applied to any product"
        # (mutually exclusive with p2pePerItemFee, out of v1 UI scope — see
        # schemas.fees module docstring). Integrated/WebPASS/Genius are the
        # CNP product types that use tokenization; PRD-HWONB-011 §3/§5
        # documents this as a Products-derived requirement the merchant-facing
        # FeesRequest schema can't see for itself — enforced here, before any
        # GP call, same as the Genius-plan check above.
        CNP_PRODUCT_TYPES = {"integrated", "webpass", "genius"}
        if (products_data or {}).get("productType") in CNP_PRODUCT_TYPES and not (
            payload.get("dataProtectionPerItemFee") or payload.get("p2pePerItemFee")
        ):
            return _fail(
                tsys_errors=[{
                    "field": "dataProtectionPerItemFee",
                    "message": "A data protection per-item fee is required because your "
                               "product selection uses tokenization.",
                }]
            )

        partner = settings.TSYS_BOARDING_PARTNER
        association = association or settings.TSYS_BOARDING_ASSOCIATION

        try:
            rate_card = await self._resolve_rate_card(
                processing_type=processing_type,
                pricing_plan_id=pricing_plan_id,
                option_id=option_id,
                partner=partner,
                association=association,
                amex_pricing_ind=amex_pricing_ind,
                business_data=business_data,
                card_types_data=card_types_data,
                named_misc_fee_overrides=named_misc_fee_overrides,
                merchant_id=merchant_id,
                application_id=application_id,
            )
        except RuntimeError as exc:
            logger.error("push_fees: rate-card resolution failed for app_id=%s — %s", app_id, exc)
            return _fail(
                tsys_errors=[{"field": "_system", "message": "Unable to resolve the pricing plan's rate card."}]
            )
        except MiscFeeRangeError as exc:
            return _fail(tsys_errors=exc.violations)

        body: Dict[str, Any] = dict(payload)

        body["baseRates"] = rate_card["baseRates"]
        body["perItemFees"] = rate_card["perItemFees"]
        body["authorizationFees"] = rate_card["authorizationFees"]
        body["pinDebitFees"] = rate_card["pinDebitFees"]
        body["miscellaneousFees"] = rate_card["miscellaneousFees"]

        # additionalFees nests both the merchant-selected fields
        # (earlyTermination, includeDailyDiscount, sameDayACHFlag/-Fee,
        # ach*, wireless*, unsupportedPos*, p2pe/dataProtection,
        # geniusPlan/geniusAddOns) AND the server-injected EnsureBill fields
        # — GP's real top-level shape nests all of these under
        # `additionalFees` (PRD-011 §1.6), while FeesRequest.to_gp_payload()
        # flattens them (matching the prototype's flat data-bind
        # convention, PRD-011 §4A). Move them under `additionalFees` here;
        # processingType/pricingPlanId/optionId/amexPricingInd/
        # transFreedomBundle stay at the top level (§1.1).
        top_level_keys = {"processingType", "pricingPlanId", "optionId", "amexPricingInd", "transFreedomBundle"}
        rate_card_keys = {"baseRates", "perItemFees", "authorizationFees", "pinDebitFees", "miscellaneousFees"}
        additional_fees = {
            k: v for k, v in body.items() if k not in top_level_keys and k not in rate_card_keys
        }
        for k in list(additional_fees.keys()):
            body.pop(k)
        # EnsureBill's setUpFee/monthlyFee/deliveryFee are FLAT siblings
        # under `additionalFees` — see `_resolve_ensure_bill`'s docstring
        # and the doc's own Sample Request/Response (pp. 324-328), which
        # show no `ensureBill` wrapper object anywhere. Merged in (not
        # sent at all) when the partner isn't configured for EnsureBill.
        if rate_card.get("ensureBill"):
            additional_fees.update(rate_card["ensureBill"])
        body["additionalFees"] = additional_fees

        envelope = await self._request(
            "POST", f"/applications/{app_id}/fees", json=body, merchant_id=merchant_id, application_id=application_id
        )
        # Extra (undeclared-in-TypedDict but present at runtime) key so
        # services.py can snapshot exactly what was sent to GP for
        # support/audit purposes (PRD-HWONB-011 §3) without re-deriving or
        # re-calling GP.
        envelope["sent_payload"] = body
        return envelope

    async def get_fee_processing_detail(
        self, partner: str, association: Optional[str] = None, processing_type: Optional[str] = None,
        merchant_id: Optional[int] = None, application_id: Optional[int] = None,
    ) -> BoardingEnvelope:
        """
        GET /feeSchedule/feeProcessingDetail — PRD-HWONB-011 §2, shape
        confirmed C-02: `{feeProcessingDetail:[{processingType,pricingPlans:
        [{pricingPlanId,description,options:[{optionId,description,
        baseRatesAllCardTypesReq,baseRatesThresholds:{baseRatePercentageFields:
        [{fieldName,min,max,defaultValue}],baseRatePerItemFields:[...]}}]}]}]}`.

        `association` defaults from settings when the generic
        `GET /onboarding/lookups/{type}` router path calls this with only
        `partner` (router.py currently only collects `partner` for every
        catalog lookup). `processing_type` narrows the GP-side filter when
        known; omit it to fetch the full catalog (used by
        `_resolve_rate_card`, which then filters client-side anyway).
        """
        params: Dict[str, str] = {
            "partner": partner,
            "association": association or settings.TSYS_BOARDING_ASSOCIATION,
        }
        if processing_type:
            params["processingType"] = processing_type
        return await self._request("GET", "/feeSchedule/feeProcessingDetail", params=params, merchant_id=merchant_id, application_id=application_id)

    async def get_fee_detail(
        self,
        partner: str,
        association: Optional[str] = None,
        fee_type: Optional[str] = None,
        fee_code: Optional[str] = None,
        merchant_id: Optional[int] = None, application_id: Optional[int] = None,
    ) -> BoardingEnvelope:
        """
        GET /feeSchedule/feeDetail (+ /feeType/{type}, +
        /feeType/miscFee/feeCode/{feeCode}) — PRD-HWONB-011 §2. No saved
        example response exists for any of these (per the PRD), so
        `_resolve_rate_card`'s callers parse the result defensively.

        `fee_type` narrows to one of {miscFee, authorizationFee,
        pinDebitFee, additionalServiceFee}; `fee_code` (only meaningful
        with fee_type="miscFee") narrows to one specific fee code's detail.
        """
        path = "/feeSchedule/feeDetail"
        if fee_type:
            path += f"/feeType/{fee_type}"
            if fee_type == "miscFee" and fee_code:
                path += f"/feeCode/{fee_code}"
        params: Dict[str, str] = {
            "partner": partner,
            "association": association or settings.TSYS_BOARDING_ASSOCIATION,
        }
        return await self._request("GET", path, params=params, merchant_id=merchant_id, application_id=application_id)

    async def get_plans_and_addons(
        self, partner: str, association: Optional[str] = None, merchant_id: Optional[int] = None, application_id: Optional[int] = None
    ) -> BoardingEnvelope:
        """
        GET /feeSchedule/plans/addons — PRD-HWONB-011 §2/§7, CERT-corrected
        path (C-04): lowercase, NOT PascalCase `/feeSchedule/PlansAndAddOns`
        (that variant 400s with `40252 INVALID_ACTION`). Shape:
        `{plansAndAddOns:[{planName,planDescription,addonsDetails:
        [{addOnFeeName,addOnDescription}]}]}`. Genius add-ons are nested in
        `addonsDetails` here — there is no separate Genius add-ons endpoint.
        """
        params: Dict[str, str] = {
            "partner": partner,
            "association": association or settings.TSYS_BOARDING_ASSOCIATION,
        }
        return await self._request("GET", fm.PLANS_AND_ADDONS_PATH, params=params, merchant_id=merchant_id, application_id=application_id)

    async def get_addons_catalog(
        self, partner: str, association: Optional[str] = None, merchant_id: Optional[int] = None, application_id: Optional[int] = None
    ) -> BoardingEnvelope:
        """
        GET /product/addOns — PRD-HWONB-011 §2. This is the STANDARD
        (non-Genius) add-on catalog: ACH terms, EnsureBill, PIN-debit
        defaults — distinct from `get_plans_and_addons()` (Genius plans +
        nested Genius add-ons, C-04). Documented (not CERT-confirmed live)
        shape: `{ach{transactionFee,returnFee,discountRate,fraudCheckFee,
        fees,settlement}, ensureBill{monthlyFee,setupFee}, pinDebit{...}}`.
        `_resolve_ensure_bill` reads `ensureBill` from here defensively.
        """
        params: Dict[str, str] = {
            "partner": partner,
            "association": association or settings.TSYS_BOARDING_ASSOCIATION,
        }
        return await self._request("GET", "/product/addOns", params=params, merchant_id=merchant_id, application_id=application_id)

    # ------------------------------------------------------------------
    # PRD-HWONB-011 §6 / C-01 — rate-card resolution (AC-14 hard blocker)
    # ------------------------------------------------------------------

    def _rate_card_cache_key(
        self, partner: str, association: str, processing_type: str, pricing_plan_id: Any, option_id: Any
    ) -> str:
        return _RATE_CARD_CACHE_KEY_TMPL.format(
            env=self._env,
            partner=partner,
            association=association,
            processing_type=processing_type,
            pricing_plan_id=pricing_plan_id,
            option_id=option_id,
        )

    @staticmethod
    def _bump_above_zero(value: Optional[Any], field_min: Optional[Any] = None) -> float:
        """
        C-01 round 2 — GP rejects any rate-card value of exactly 0.00
        (error 40213). Use the field's own `defaultValue` when it's
        already positive; otherwise fall back to a positive `min` if GP
        supplied one, and only fall back to the hardcoded
        `RATE_CARD_ZERO_FLOOR` when neither is usable. Every observed
        `min` in the one CERT-confirmed example was 0, so the floor
        constant is expected to be the common case in practice.
        """
        try:
            numeric = float(value) if value is not None else 0.0
        except (TypeError, ValueError):
            numeric = 0.0
        if numeric > 0:
            return numeric
        try:
            min_numeric = float(field_min) if field_min is not None else 0.0
        except (TypeError, ValueError):
            min_numeric = 0.0
        return min_numeric if min_numeric > 0 else fm.RATE_CARD_ZERO_FLOOR

    @staticmethod
    def _is_trans_freedom_description(description: Optional[str]) -> bool:
        """
        GP flags certain pricing plan/options as TransFreedom via a plain
        `description` string on the feeProcessingDetail catalog node — no
        dedicated boolean field exists. Substring match (not exact), since
        a live CERT trial (2026-07-20) showed a real association renamed
        its display name ("AC148 - Cash Advance") while its underlying
        plan/option `description` values stayed exactly "TransFreedom" —
        exact-match would miss variant casing/whitespace GP has been
        observed to use ("TRANSFREEDOM ").
        """
        return "transfreedom" in str(description or "").strip().lower()

    @staticmethod
    def _find_fee_processing_option(
        data: Any, processing_type: str, pricing_plan_id: Any, option_id: Any
    ) -> Optional[Dict[str, Any]]:
        """Walk the feeProcessingDetail response (C-02 shape) to the matching (processingType, pricingPlanId, optionId) option node."""
        if not isinstance(data, dict):
            return None
        for pt_entry in data.get("feeProcessingDetail") or []:
            if pt_entry.get("processingType") != processing_type:
                continue
            for plan in pt_entry.get("pricingPlans") or []:
                if str(plan.get("pricingPlanId")) != str(pricing_plan_id):
                    continue
                for option in plan.get("options") or []:
                    if str(option.get("optionId")) == str(option_id):
                        return option
        return None

    def _inject_threshold_field(
        self,
        field: Dict[str, Any],
        suffix: str,
        field_map: Dict[str, Tuple[str, str]],
        target: Dict[str, float],
        index: int,
    ) -> None:
        """
        Map one `baseRatePercentageFields`/`baseRatePerItemFields` entry
        (`{fieldName, min, max, defaultValue}`) onto its baseRates/
        perItemFees payload key via RATE_CARD_FIELD_NAME_MAP, bumping a
        zero/None defaultValue above zero (C-01 round 2). Unrecognized
        `fieldName` stems are logged and skipped rather than guessed —
        see the WARNING on RATE_CARD_FIELD_NAME_MAP in field_mapping.py.
        """
        field_name = field.get("fieldName") or ""
        stem = field_name[: -len(suffix)] if field_name.endswith(suffix) else field_name
        mapping = field_map.get(stem)
        if not mapping:
            logger.warning(
                "_resolve_rate_card: unrecognized rate field %r (stem %r) — no entry in "
                "RATE_CARD_FIELD_NAME_MAP, skipping. Needs a CERT trial against a plan "
                "that exercises this field to confirm the correct baseRates/perItemFees key.",
                field_name,
                stem,
            )
            return
        key = mapping[index]
        target[key] = self._bump_above_zero(field.get("defaultValue"), field.get("min"))

    @staticmethod
    def _extract_fee_catalog_defaults(envelope: BoardingEnvelope, keys: Tuple[str, ...]) -> Dict[str, Any]:
        """
        Defensively parse a `GET /feeSchedule/feeDetail/feeType/{type}`
        response into `{wanted_key: defaultValue}`. This accepts either
        `{<something>: [...]}` or a bare list, matching each entry's
        `feeName`/`fieldName`/`feeCode`/`name` case-insensitively against
        `keys`. `feeName`/`feeDefaultValue` is the live-CERT-confirmed
        shape (2026-07-20) for `additionalServiceFee` entries (e.g.
        `{"feeName": "EnsureBILLSetupFee", "feeDefaultValue": 0}`) — checked
        first; `fieldName`/`defaultValue` stays as a fallback for any
        catalog that uses that shape instead.
        """
        if not envelope.get("ok"):
            return {}
        data = envelope.get("data")
        entries: List[Dict[str, Any]] = []
        if isinstance(data, dict):
            for v in data.values():
                if isinstance(v, list):
                    entries.extend(v)
        elif isinstance(data, list):
            entries = data

        lookup = {str(k).lower(): k for k in keys}
        result: Dict[str, Any] = {}
        for entry in entries:
            if not isinstance(entry, dict):
                continue
            name = str(
                entry.get("feeName") or entry.get("fieldName") or entry.get("feeCode") or entry.get("name") or ""
            ).lower()
            if name in lookup:
                result[lookup[name]] = entry.get("feeDefaultValue", entry.get("defaultValue"))
        return result

    async def _resolve_authorization_fees(
        self, partner: str, association: str, merchant_id: Optional[int] = None, application_id: Optional[int] = None
    ) -> Dict[str, float]:
        """
        §1.3 authorizationFees — unconditionally-required object with 5
        fields (allCardTypes/batchClose/voice/aru/amex). No CERT-confirmed
        example response exists for
        `GET /feeSchedule/feeDetail/feeType/authorizationFee` (PRD-011 §2 —
        "No saved example response"), so this defensively parses whatever
        shape comes back and falls back to the zero-floor default per
        field if the catalog is empty/unrecognized. NEEDS a real CERT call
        to confirm the actual response shape before trusting these as more
        than "GP will accept it", not necessarily the real negotiated rate.
        """
        envelope = await self.get_fee_detail(
            partner, association, fee_type="authorizationFee", merchant_id=merchant_id, application_id=application_id
        )
        catalog = self._extract_fee_catalog_defaults(envelope, fm.AUTHORIZATION_FEE_KEYS)
        return {key: self._bump_above_zero(catalog.get(key)) for key in fm.AUTHORIZATION_FEE_KEYS}

    @staticmethod
    def _unwrap_pin_debit_plan(data: Any) -> Optional[Dict[str, Any]]:
        """
        `GET /feeSchedule/feeDetail/feeType/pinDebitFee` is documented (PDF
        pp.44-47) as returning a flat `{pricingPlanId, description,
        plans:[...]}` object, but the CERT sandbox's actual live response
        instead wraps that same object under a stringified numeric index key
        (`{"0": {pricingPlanId: ..., ...}, "action": {...}}` — an apparent
        GP quirk turning a single-element array into an object). Accept
        either shape defensively rather than trusting the documented one.
        """
        if not isinstance(data, dict):
            return None
        if "pricingPlanId" in data:
            return data
        for key, value in data.items():
            if key == "action" or not isinstance(value, dict):
                continue
            if "pricingPlanId" in value:
                return value
        return None

    async def _resolve_pin_debit_fees(
        self, partner: str, association: str, card_types_data: Optional[Dict[str, Any]] = None,
        merchant_id: Optional[int] = None, application_id: Optional[int] = None,
    ) -> Dict[str, Any]:
        """
        §1.4 pinDebitFees. Per the PDF (pp.317/341), `pdpPricePlan` and the
        5 rate fields are only Cond-Mand when `cardTypes.debitRequested=true`
        (`pinDebitEBTPerItemFee` on `ebtRequested` instead) — but a live CERT
        trial (this merchant, debitRequested=false) showed GP rejects
        `pdpPricePlan: null` with "Pin Debit Pricing Plan details not found
        for the given pricing plan id" regardless of debitRequested. So
        `pdpPricePlan` is resolved from the partner/association-level
        catalog unconditionally (it's a partner config value, not a
        merchant opt-in, and costs nothing extra to look up); only the 5
        rate/amount sub-fields stay gated on debitRequested/ebtRequested,
        matching GP's own "MOTO - All Null" example for those.

        See `_unwrap_pin_debit_plan` for the real (undocumented) response
        shape this catalog call returns.
        """
        debit_requested = bool((card_types_data or {}).get("debitRequested"))
        ebt_requested = bool((card_types_data or {}).get("ebtRequested"))

        result: Dict[str, Any] = {"pdpPricePlan": None}
        result.update({key: None for key in fm.PIN_DEBIT_FEE_RATE_KEYS})

        envelope = await self.get_fee_detail(
            partner, association, fee_type="pinDebitFee", merchant_id=merchant_id, application_id=application_id
        )
        plan = self._unwrap_pin_debit_plan(envelope.get("data")) if envelope.get("ok") else None
        plan_id = (plan or {}).get("pricingPlanId")

        if plan_id is None and debit_requested:
            raise RuntimeError(
                f"pinDebitFee lookup for partner={partner} association={association} "
                "returned no pricingPlanId — cardTypes.debitRequested is true but GP has "
                "no pin-debit pricing plan configured for this partner/association."
            )
        result["pdpPricePlan"] = plan_id

        if debit_requested:
            # The 5 rate sub-fields' real values live in `plan["plans"][].
            # elementName`/`rval` ranges with no CERT-confirmed field mapping
            # (PRD-011 §2 — "No saved example response"); keep the existing
            # zero-floor default here until a real CERT trial confirms how
            # to map them — see push_fees()'s CERT-flakiness note.
            catalog = self._extract_fee_catalog_defaults(envelope, fm.PIN_DEBIT_FEE_RATE_KEYS)
            for key in fm.PIN_DEBIT_FEE_RATE_KEYS:
                if key == "pinDebitEBTPerItemFee":
                    continue
                result[key] = self._bump_above_zero(catalog.get(key))

        if ebt_requested:
            result["pinDebitEBTPerItemFee"] = self._bump_above_zero(None)

        return result

    def _build_misc_fee_entry(
        self,
        fee_code: str,
        amount: Any,
        frequency: str,
        seasonal: bool,
        business_data: Optional[Dict[str, Any]],
    ) -> Dict[str, Any]:
        """
        One `miscellaneousFees[]` entry, month-flags set per §1.5's
        frequency bucketing: Seasonal-Monthly mirrors the business's own
        operating-months flags (GP: "seasonal fee selected month must match
        ... business configuration page"); every other frequency
        (Annual/One-Time/Per-Instance/plain Monthly) sets only the
        `feeStartDate`'s own month flag true. Only the Seasonal-Monthly
        case (feeCode 8001) is CERT-confirmed — the rest is this same
        documented pattern extrapolated to the newly-exposed named fees,
        since no CERT trial has exercised them yet. Callers must NOT derive
        `seasonal` for feeCode 8001 from the fee catalog's own `seasonalFee`
        flag — GP's catalog can say `False` while GP's own `/fees`
        submission validator still enforces the mirror for this code
        (CERT-confirmed 2026-07-28); see `_resolve_misc_fees`.
        """
        today = _dt.date.today()
        entry: Dict[str, Any] = {
            "feeStartDate": today.isoformat(),
            "chargeAmount": amount,
            "feeCode": fee_code,
        }
        additional_info = (business_data or {}).get("additionalInfo") or {}
        if seasonal and frequency == "Monthly":
            for fee_flag_key, business_flag_key in zip(fm.MISC_FEE_MONTH_FLAG_KEYS, fm.BUSINESS_MONTH_FLAG_KEYS):
                entry[fee_flag_key] = bool(additional_info.get(business_flag_key, True))
        else:
            start_month_index = today.month - 1
            for index, flag_key in enumerate(fm.MISC_FEE_MONTH_FLAG_KEYS):
                entry[flag_key] = index == start_month_index
        return entry

    async def _resolve_misc_fees(
        self,
        partner: str,
        association: str,
        business_data: Optional[Dict[str, Any]] = None,
        named_fee_overrides: Optional[Dict[str, float]] = None,
        merchant_id: Optional[int] = None, application_id: Optional[int] = None,
    ) -> List[Dict[str, Any]]:
        """
        §1.5 miscellaneousFees — CERT-confirmed MANDATORY and non-empty
        (both `[]` and `null` → 40251 "cannot be empty"). Builds one entry
        per named fee the merchant explicitly opted into
        (`NAMED_MISC_FEE_CODES`, keyed by `FeesRequest`'s
        `named_misc_fee_overrides()`), plus always the plan's default
        Monthly Service Fee entry (feeCode 8001, the CERT-confirmed
        example) so GP always receives >=1 entry even when the merchant
        left every named fee blank — falling back to the hardcoded
        `FALLBACK_MISC_FEE_CODE`/`FALLBACK_MISC_FEE_AMOUNT` if the catalog
        call fails or returns nothing usable for 8001.
        """
        overrides = named_fee_overrides or {}

        catalog_by_code: Dict[str, Dict[str, Any]] = {}
        envelope = await self.get_fee_detail(
            partner, association, fee_type="miscFee", merchant_id=merchant_id, application_id=application_id
        )
        if envelope.get("ok"):
            data = envelope.get("data")
            misc_list = (data or {}).get("miscFee") if isinstance(data, dict) else None
            if isinstance(misc_list, list):
                for row in misc_list:
                    code = str(row.get("feeCode", ""))
                    if code:
                        catalog_by_code[code] = row

        entries: List[Dict[str, Any]] = []
        violations: List[TsysError] = []
        for alias, (fee_code, frequency) in fm.NAMED_MISC_FEE_CODES.items():
            catalog_row = catalog_by_code.get(fee_code, {})
            is_monthly_service_fee = fee_code == fm.FALLBACK_MISC_FEE_CODE

            merchant_amount = overrides.get(alias)
            if merchant_amount is None and not is_monthly_service_fee:
                continue  # Optional named fee the merchant didn't opt into.

            if merchant_amount is not None:
                # p.317/341 — chargeAmount "will be determined off of the
                # feeMin and feeMax amount returned in the Master Record
                # Miscellaneous request" (this same catalog_row). Reject a
                # merchant override outside that range here — the trust
                # boundary — instead of letting GP bounce it back raw.
                fee_min = catalog_row.get("feeMin", 0)
                fee_max = catalog_row.get("feeMax")
                if fee_max is not None and not (fee_min <= merchant_amount <= fee_max):
                    violations.append({
                        "field": alias,
                        "message": f"{alias} must be between ${fee_min:.2f} and ${fee_max:.2f}",
                    })
                    continue
                amount: Any = merchant_amount
            else:
                # GP's catalog genuinely returns feeDefaultValue: null for 8001
                # (no plan default exists) — `.get(key, default)` only falls
                # back on a missing key, not an explicit None, so that has to
                # be checked here rather than via the dict default arg.
                default_amount = catalog_row.get("feeDefaultValue")
                amount = default_amount if default_amount is not None else fm.FALLBACK_MISC_FEE_AMOUNT

            if is_monthly_service_fee:
                # GP's live fee catalog can report seasonalFee: False for
                # feeCode 8001 on some associations (CERT-confirmed
                # 2026-07-28, association 127806, tsys_app_id=2002169) — but
                # GP's own /fees submission validator still enforces the
                # seasonal business-month mirror for this fee regardless,
                # rejecting a non-mirrored entry with "seasonal fee ... must
                # match ... business configuration page" (40256). Never
                # trust the catalog for this one well-known code; always
                # mirror business_data's month flags.
                seasonal, resolved_frequency = True, "Monthly"
            else:
                seasonal = bool(catalog_row.get("seasonalFee", False))
                resolved_frequency = catalog_row.get("feeFrequency", frequency)
            entries.append(
                self._build_misc_fee_entry(fee_code, amount, resolved_frequency, seasonal, business_data)
            )

        if violations:
            raise MiscFeeRangeError(violations)

        if not entries:
            # Catalog lookup returned nothing at all for 8001 and it wasn't
            # in NAMED_MISC_FEE_CODES's loop output — pure defensive
            # fallback, should be unreachable since monthlyServiceFee is
            # always injected above regardless of overrides.
            entries.append(
                self._build_misc_fee_entry(
                    fm.FALLBACK_MISC_FEE_CODE, fm.FALLBACK_MISC_FEE_AMOUNT, "Monthly", True, business_data
                )
            )

        return entries

    async def _resolve_ensure_bill(
        self, partner: str, association: str, merchant_id: Optional[int] = None, application_id: Optional[int] = None
    ) -> Optional[Dict[str, float]]:
        """
        EnsureBill's `setUpFee`/`monthlyFee`/`deliveryFee` are FLAT sibling
        keys directly under `additionalFees` — GP's own Base Boarding API
        Developer Guide PDF documents the field table as
        `additionalFees.ensureBill.X`, but the SAME doc's Sample
        Request/Response (pp. 324-328) show no `ensureBill` wrapper object
        at all, just `additionalFees.setUpFee`/`.monthlyFee`/`.deliveryFee`
        directly — confirmed authoritative here since a live sample is what
        GP's parser actually accepts, not a field-table label. This method
        still returns them as one small dict for `push_fees()` to fold
        (flatten) into `additionalFees`; it is NOT sent as a nested object.
        The PDF also documents "0.00 or null allowed" for all three, so no
        zero-floor bump is needed here.

        "Not all Client/Partners are setup with EnsureBill" (PDF p.322) —
        `GET /feeSchedule/feeDetail/feeType/additionalServiceFee` is the
        confirmed-live source for that (GP's own Postman "Starter
        Collection"; `GET /product/addOns`'s `ensureBill` sub-object was
        tried first but confirmed unreliable — it returns non-empty
        regardless of the partner's real EnsureBill enablement). Return
        `None` (omit the object entirely) when the catalog doesn't confirm
        EnsureBill at all, rather than fabricating fees for a merchant not
        configured for it.
        """
        envelope = await self.get_fee_detail(
            partner, association, fee_type="additionalServiceFee", merchant_id=merchant_id, application_id=application_id
        )
        catalog = self._extract_fee_catalog_defaults(
            envelope,
            (fm.ENSURE_BILL_SETUP_FEE_CATALOG_KEY, fm.ENSURE_BILL_MONTHLY_FEE_CATALOG_KEY),
        )

        if not catalog:
            return None

        return {
            "setUpFee": catalog.get(fm.ENSURE_BILL_SETUP_FEE_CATALOG_KEY) or 0,
            "monthlyFee": catalog.get(fm.ENSURE_BILL_MONTHLY_FEE_CATALOG_KEY) or 0,
            # No catalog source confirmed for deliveryFee — 0 is a documented-valid default.
            "deliveryFee": 0,
        }

    @staticmethod
    def _apply_amex_gate(resolved: Dict[str, Any], amex_pricing_ind: bool) -> Dict[str, Any]:
        """
        C-01 — `amexBaseRate`/`amexPerItem` are disallowed (error 40008)
        when `amexPricingInd=false`; omit them entirely in that case.
        Operates on a deep copy so the Redis-cached resolved rate card
        (which is NOT keyed by amexPricingInd) is never mutated in place.
        """
        result = copy.deepcopy(resolved)
        if not amex_pricing_ind:
            result["baseRates"].pop("amexBaseRate", None)
            result["perItemFees"].pop("amexPerItem", None)
        return result

    @staticmethod
    def _apply_trans_freedom_zero_fill(resolved: Dict[str, Any]) -> Dict[str, Any]:
        """
        Base Boarding API Guide pp. 314-317 — when the selected plan/option
        is TransFreedom-classified, GP requires this specific set of fields
        sent as 0.00 rather than their normal negotiated/catalog values.
        This is the documented behavior for TransFreedom, NOT a reason to
        block the submission (an earlier version of `push_fees()` returned
        a `_system` error instead — wrong, since no such
        block/transFreedomType field exists anywhere in the docs).

        `amexBaseRate`/`amexPerItem` are set here unconditionally but only
        survive if `_apply_amex_gate` (always applied right after this)
        keeps them — i.e. only when amexPricingInd is true, matching the
        doc's "...and amexPricingInd equals true, then amexBaseRate is
        required to be 0.00". Operates on a deep copy, mirroring
        `_apply_amex_gate`.
        """
        if not resolved.get("isTransFreedom"):
            return resolved
        result = copy.deepcopy(resolved)
        result["baseRates"]["actBaseRateOrVsMcBaseRate"] = 0.0
        result["baseRates"]["amexBaseRate"] = 0.0
        result["perItemFees"]["actPerItemOrVsMcPerItem"] = 0.0
        result["perItemFees"]["amexPerItem"] = 0.0
        for key in ("allCardTypes", "batchClose", "voice", "aru"):
            result["authorizationFees"][key] = 0.0
        for key in ("pinDebitPerItemFee", "pinDebitRatePercent", "pinDebitEBTPerItemFee"):
            result["pinDebitFees"][key] = 0.0
        return result

    async def _resolve_rate_card(
        self,
        processing_type: str,
        pricing_plan_id: Any,
        option_id: Any,
        partner: str,
        association: str,
        *,
        amex_pricing_ind: bool,
        business_data: Optional[Dict[str, Any]] = None,
        card_types_data: Optional[Dict[str, Any]] = None,
        named_misc_fee_overrides: Optional[Dict[str, float]] = None,
        merchant_id: Optional[int] = None, application_id: Optional[int] = None,
    ) -> Dict[str, Any]:
        """
        PRD-HWONB-011 §6 / PRD-HWONB-001 §13.1 C-01 — the AC-14 hard
        blocker. Server-side injects the rate-card fields GP requires
        non-null (`baseRates`, `perItemFees`, `authorizationFees`,
        `pinDebitFees`, `miscellaneousFees`, `additionalFees.ensureBill`)
        for the merchant's chosen (processingType, pricingPlanId, optionId)
        instead of asking the merchant to type in a partner-negotiated
        rate card. Called from `push_fees()` only — this is a private
        resolver, not a router-facing method.

        ================================================================
        THIS IS THE HIGHEST-RISK CODE PATH IN THE ONBOARDING MODULE.
        ================================================================
        CERT (2026-07-14) confirmed the *rules* GP enforces field-by-field
        (rates must be >0.00 not 0.00 — 40213; amex fields must be omitted
        entirely when amexPricingInd=false — 40008; miscellaneousFees must
        be non-empty with a real feeCode; pinDebitFees rates must be >0) —
        see C-01 round 2 in PRD-HWONB-001 §13.1. A *fully clean* 201 on
        plan-30/option-41 "TransFreedom" was NOT reached in that same CERT
        session: CERT returned inconsistent validation errors across
        byte-identical request bodies for the TEST833 test partner
        (ensureBill/pinDebit rules appeared to toggle between passes).
        Root-caused 2026-07-20 by reading the Base Boarding API Guide in
        full (pp. 313-328): it wasn't flakiness, it was two real bugs —
        `ensureBill` was sent as a nested object GP doesn't look for (see
        `_resolve_ensure_bill`'s docstring), and TransFreedom was treated
        as unsendable instead of zero-filled (see
        `_apply_trans_freedom_zero_fill`). Both are now fixed below.
        ================================================================

        Resolution steps:
          1. `baseRates`/`perItemFees` — from
             `get_fee_processing_detail()`'s matching option's
             `baseRatesThresholds`, mapped field-by-field via
             RATE_CARD_FIELD_NAME_MAP (only the "AllCardTypes" stem is
             CERT-confirmed; see the WARNING on that map).
          2. `authorizationFees`/`pinDebitFees` — from
             `get_fee_detail(fee_type=...)` catalogs, defensively parsed
             (no CERT-confirmed example response exists for either).
          3. `miscellaneousFees` — from `get_fee_detail(fee_type="miscFee")`,
             falling back to the CERT-confirmed 8001 "MONTHLY SERVICE FEE".
          4. `ensureBill` — from `get_fee_detail(fee_type="additionalServiceFee")`,
             returned flat (`_resolve_ensure_bill`) for `push_fees()` to fold
             into `additionalFees` — never sent as a nested object.
          5. Every resolved rate is bumped above 0.00 (`_bump_above_zero`).
          6. If the plan/option is TransFreedom-classified, the doc-mandated
             set of fields is forced to 0.00 (`_apply_trans_freedom_zero_fill`)
             — applied to both the cache-hit and cache-miss paths below,
             since it's a per-request overlay, not part of the cached
             partner/plan-negotiated data.
          7. `amexBaseRate`/`amexPerItem` are included/excluded per
             `amex_pricing_ind` (`_apply_amex_gate`) — applied LAST (after
             step 6), since the cache key isn't scoped by amexPricingInd
             (the rest of the rate card doesn't vary by it).

        Cached in Redis for `_RATE_CARD_CACHE_TTL_SECONDS` (1 day) keyed by
        (env, partner, association, processingType, pricingPlanId, optionId)
        — this data is partner+association-negotiated and changes rarely, so
        `push_fees()` doesn't need to re-hit 3-4 GP lookups on every save.
        Mirrors the token Redis-cache pattern in `mint_token()`/`_get_token()`
        above.
        """
        # `pinDebitFees` (gated on this merchant's cardTypes.debitRequested/
        # ebtRequested) and `miscellaneousFees` (mirrors this merchant's own
        # Business Information month flags) are merchant-specific, not
        # partner+plan-negotiated — they must NOT be cached under a key
        # scoped only by (partner, association, processingType,
        # pricingPlanId, optionId), or one merchant's debit opt-in / seasonal
        # months would leak into another merchant's identical plan
        # selection. Resolve them fresh on every call, same as the
        # post-cache amex gating below.
        pin_debit_fees = await self._resolve_pin_debit_fees(
            partner, association, card_types_data, merchant_id=merchant_id, application_id=application_id
        )
        misc_fees = await self._resolve_misc_fees(
            partner, association, business_data, named_misc_fee_overrides, merchant_id=merchant_id, application_id=application_id
        )

        cache_key = self._rate_card_cache_key(partner, association, processing_type, pricing_plan_id, option_id)
        r = _get_redis_client()
        if r is not None:
            try:
                cached_raw = r.get(cache_key)
                if cached_raw:
                    cached = json.loads(cached_raw)
                    cached.setdefault("isTransFreedom", False)
                    cached["pinDebitFees"] = pin_debit_fees
                    cached["miscellaneousFees"] = misc_fees
                    zero_filled = self._apply_trans_freedom_zero_fill(cached)
                    return self._apply_amex_gate(zero_filled, amex_pricing_ind)
            except Exception as exc:
                logger.warning("_resolve_rate_card: cache read failed — %s", exc)

        # 1) baseRates / perItemFees from feeProcessingDetail thresholds (C-02).
        # GP CERT 400s on the processingType query filter here ("processingType
        # value is invalid") even though the generic lookups/feeProcessingDetail
        # dropdown endpoint accepts it fine unfiltered — omit it and rely on
        # _find_fee_processing_option's existing client-side filtering below.
        fpd_envelope = await self.get_fee_processing_detail(partner, association, merchant_id=merchant_id, application_id=application_id)
        if not fpd_envelope["ok"]:
            raise RuntimeError(
                f"feeProcessingDetail lookup failed while resolving rate card: {fpd_envelope['tsys_errors']}"
            )

        option_node = self._find_fee_processing_option(
            fpd_envelope["data"], processing_type, pricing_plan_id, option_id
        )
        base_rates: Dict[str, float] = {}
        per_item_fees: Dict[str, float] = {}
        if option_node:
            thresholds = option_node.get("baseRatesThresholds") or {}
            for pct_field in thresholds.get("baseRatePercentageFields") or []:
                self._inject_threshold_field(
                    pct_field, "Percentage", fm.RATE_CARD_FIELD_NAME_MAP, base_rates, index=0
                )
            for pi_field in thresholds.get("baseRatePerItemFields") or []:
                self._inject_threshold_field(
                    pi_field, "PerItem", fm.RATE_CARD_FIELD_NAME_MAP, per_item_fees, index=1
                )
        else:
            logger.warning(
                "_resolve_rate_card: no matching (processingType=%s, pricingPlanId=%s, "
                "optionId=%s) found in feeProcessingDetail response — falling back to the "
                "single CERT-confirmed AllCardTypes bucket at the zero-floor default. This "
                "WILL need a real CERT trial against the merchant's actual plan to confirm "
                "correctness before transmit.",
                processing_type,
                pricing_plan_id,
                option_id,
            )
            base_key, per_key = fm.RATE_CARD_FIELD_NAME_MAP["AllCardTypes"]
            base_rates[base_key] = fm.RATE_CARD_ZERO_FLOOR
            per_item_fees[per_key] = fm.RATE_CARD_ZERO_FLOOR

        is_trans_freedom = bool(option_node) and self._is_trans_freedom_description(
            option_node.get("description")
        )

        # 2) authorizationFees — partner+plan negotiated, cacheable.
        authorization_fees = await self._resolve_authorization_fees(partner, association, merchant_id=merchant_id, application_id=application_id)

        # 3) additionalFees ensureBill fields — partner-configured, cacheable.
        ensure_bill = await self._resolve_ensure_bill(partner, association, merchant_id=merchant_id, application_id=application_id)

        cacheable = {
            "baseRates": base_rates,
            "perItemFees": per_item_fees,
            "authorizationFees": authorization_fees,
            "ensureBill": ensure_bill,
            "isTransFreedom": is_trans_freedom,
        }

        if r is not None:
            try:
                r.setex(cache_key, _RATE_CARD_CACHE_TTL_SECONDS, json.dumps(cacheable))
            except Exception as exc:
                logger.warning("_resolve_rate_card: cache write failed — %s", exc)

        resolved = dict(cacheable)
        resolved["pinDebitFees"] = pin_debit_fees
        resolved["miscellaneousFees"] = misc_fees

        zero_filled = self._apply_trans_freedom_zero_fill(resolved)
        return self._apply_amex_gate(zero_filled, amex_pricing_ind)

    async def association_is_viable(
        self, partner: str, association: str, merchant_id: Optional[int] = None, application_id: Optional[int] = None
    ) -> bool:
        """
        `False` only when every processingType/plan/option in this
        association's feeProcessingDetail catalog is TransFreedom-classified
        — i.e. there is no non-TransFreedom option a merchant could ever
        pick, so Fees can never complete for this association no matter
        what plan/option they choose. Callers (Application Setup's save
        handler) should reject the association up front with this, rather
        than letting the merchant discover it later as an opaque Fees-step
        failure — association *names* alone aren't a reliable signal (a
        live CERT trial, 2026-07-20, found association 086386 renamed
        "Cash Advance" while still being 100% TransFreedom underneath), so
        this does a real catalog lookup instead of a name check.

        Cached in Redis (`_ASSOCIATION_VIABILITY_CACHE_TTL_SECONDS`, 1 day)
        keyed by (env, partner, association) — same rationale as
        `_resolve_rate_card`'s cache: this is partner+association-negotiated
        data that changes rarely.
        """
        cache_key = _ASSOCIATION_VIABILITY_CACHE_KEY_TMPL.format(
            env=self._env, partner=partner, association=association
        )
        r = _get_redis_client()
        if r is not None:
            try:
                cached_raw = r.get(cache_key)
                if cached_raw is not None:
                    value = cached_raw.decode() if isinstance(cached_raw, bytes) else str(cached_raw)
                    return value == "1"
            except Exception as exc:
                logger.warning("association_is_viable: cache read failed — %s", exc)

        envelope = await self.get_fee_processing_detail(partner, association, merchant_id=merchant_id, application_id=application_id)
        viable = True
        if envelope.get("ok"):
            saw_any_option = False
            all_trans_freedom = True
            for pt_entry in (envelope.get("data") or {}).get("feeProcessingDetail") or []:
                for plan in pt_entry.get("pricingPlans") or []:
                    for option in plan.get("options") or []:
                        saw_any_option = True
                        if not self._is_trans_freedom_description(option.get("description")):
                            all_trans_freedom = False
            if saw_any_option:
                viable = not all_trans_freedom

        if r is not None:
            try:
                r.setex(cache_key, _ASSOCIATION_VIABILITY_CACHE_TTL_SECONDS, "1" if viable else "0")
            except Exception as exc:
                logger.warning("association_is_viable: cache write failed — %s", exc)

        return viable

    async def get_uma(self, app_id: int, merchant_id: Optional[int] = None, application_id: Optional[int] = None) -> BoardingEnvelope:
        """
        GET /applications/{appId}/uma — PRD-HWONB-012 §1.

        Resolved plural path (the PDF's singular `application/{appId}/uma` is a
        documented typo — every real Postman example uses the plural form).
        Idempotent/re-callable — GP returns the current state of the agreement
        every time, not a single-use generator. Response is a binary PDF; the
        generic `_request()` wrapper already returns raw bytes in `data` for any
        non-JSON content-type, so no special-casing is needed here.
        """
        return await self._request("GET", f"/applications/{app_id}/uma", merchant_id=merchant_id, application_id=application_id)

    async def upload_attachment(
        self, app_id: int, doc_type: str, file_bytes: bytes, file_name: str, content_type: str,
        merchant_id: Optional[int] = None, application_id: Optional[int] = None,
    ) -> BoardingEnvelope:
        """
        POST /applications/{appId}/attachments/upload?docType={doc_type} — PRD-HWONB-012 §2.1.

        CERT-confirmed (C-21): multipart field key MUST be `documents`, not
        `file` — sending `file` reaches GP but fails downstream validation.
        `docType` is a repeatable query param on GP's side (one call can cover
        multiple doc types for the same file); this method covers the simpler
        one-call-per-file/docType-pair case, which the PRD explicitly allows.
        """
        files = {"documents": (file_name, file_bytes, content_type)}
        return await self._request(
            "POST",
            f"/applications/{app_id}/attachments/upload",
            params={"docType": doc_type},
            files=files,
            merchant_id=merchant_id, application_id=application_id,
        )

    async def upload_attachments(
        self, app_id: int, parts: list, merchant_id: Optional[int] = None, application_id: Optional[int] = None,
    ) -> BoardingEnvelope:
        """
        Multi-document variant of upload_attachment. `parts` is a list of
        (doc_type, file_name, content_type, file_bytes) tuples; GP maps the
        repeatable `docType` query params positionally to the repeated
        `documents` file parts (e.g. `?docType=SMA&docType=CH` with two files).

        Required for charity501c3 applications: GP rejects a standalone
        `docType=SMA`/`docType=ACH` upload with 40251 "Proof of 501(c)(3)
        Status required." — even if a CH attachment already exists — and only
        accepts them when the CH (proof) file rides along in the same call.
        """
        files = [("documents", (fn, fb, ct)) for (_dt, fn, ct, fb) in parts]
        params = {"docType": [dt for (dt, _fn, _ct, _fb) in parts]}
        return await self._request(
            "POST",
            f"/applications/{app_id}/attachments/upload",
            params=params,
            files=files,
            merchant_id=merchant_id, application_id=application_id,
        )

    async def list_attachments(self, app_id: int, merchant_id: Optional[int] = None, application_id: Optional[int] = None) -> BoardingEnvelope:
        """
        GET /applications/{appId}/attachments — PRD-HWONB-012 §2.5.

        CERT-confirmed (C-22) response shape: `{attachments: [{fileName,
        attachmentId, attachmentType, fileSize, mediaType, createdAt,
        saveDraft}], action}` — the id field is `attachmentId`, not `docId`.
        """
        return await self._request("GET", f"/applications/{app_id}/attachments", merchant_id=merchant_id, application_id=application_id)

    async def delete_attachment(self, app_id: int, attachment_id: int, merchant_id: Optional[int] = None, application_id: Optional[int] = None) -> BoardingEnvelope:
        """
        DELETE /applications/{appId}/attachments/{attachmentId} — PRD-HWONB-012 §2.5.

        CERT-confirmed (C-22): 200 + the deleted attachment object plus
        `"action": {"type": "DELETE_ATTACHMENTS", "resultCode": "SUCCESS"}`.
        Pre-transmit only — services.delete_document() enforces that, not this
        thin client method.
        """
        return await self._request("DELETE", f"/applications/{app_id}/attachments/{attachment_id}", merchant_id=merchant_id, application_id=application_id)

    async def transmit(self, app_id: int, merchant_id: Optional[int] = None, application_id: Optional[int] = None) -> BoardingEnvelope:
        """
        POST /applications/{appId}/transmit — PRD-HWONB-013 §1. Point of no
        return — the caller (services.transmit()) is responsible for the local
        9-required-sections re-validation BEFORE this is ever called, so a real
        GP transmit attempt is never wasted on an application HubWallet already
        knows is incomplete (§1.3).

        Known non-error-shaped failure: a re-transmit of an already-transmitted
        appId returns `"AppId: ... is already transmitted."` in errorDetails —
        the caller treats that as an idempotent no-op success (§1.2/AC-2), not
        a hard failure. `errorCode` in any failure body is never assumed
        numeric (a real example returns the opaque string
        `"SYSTEM_ERROR_DOWNSTREAM"`); `_parse_error_body()` already handles
        that generically.
        """
        return await self._request("POST", f"/applications/{app_id}/transmit", merchant_id=merchant_id, application_id=application_id)

    async def get_status(self, app_id: int, merchant_id: Optional[int] = None, application_id: Optional[int] = None) -> BoardingEnvelope:
        """
        GET /applications/{appId}/status — PRD-HWONB-013 §2.1.

        Coarse status: `{totalApplications, ..., dashboardDTOList: [{appId,
        applicationStatus, approvalStatus, merchantBoardingStatus, mid, ...}],
        action}`. `applicationStatus`/`approvalStatus`/`merchantBoardingStatus`
        are NOT fully enumerated in any source material (C-24, still ◐
        partial as of PRD-HWONB-013) — callers must treat these as opaque
        display strings, never branch business logic off an assumed exact
        string match. `dashboardDTOList` is empty pre-transmit (CERT-observed).
        """
        return await self._request("GET", f"/applications/{app_id}/status", merchant_id=merchant_id, application_id=application_id)

    async def get_activity(self, mid: str, merchant_id: Optional[int] = None, application_id: Optional[int] = None) -> BoardingEnvelope:
        """
        GET /applications/activity?merchantId={mid} — PRD-HWONB-013 §2.2.

        The reliable, fully-enumerated endpoint (13 `activity` values / 10
        `activityType` values / 6 `status` values) — drive the progress UI off
        this, never off get_status()'s unenumerated strings (C-24). Only
        callable once a MID exists (i.e. after transmit), which is why this
        takes `mid` directly rather than `app_id` — the only caller
        (services.refresh_status) already has `application.tsys_mid` in hand
        and there is no other reference to this method's old signature to
        keep compatible (grep-verified before changing it from the initial
        `app_id`-only stub signature).
        """
        return await self._request("GET", "/applications/activity", params={"merchantId": mid}, merchant_id=merchant_id, application_id=application_id)


# Module-level singleton — stateless aside from Redis-backed token cache,
# safe to import and reuse across requests/tasks.
base_boarding_client = BaseBoardingClient()