"""
merchant_onboarding business logic.

`start_or_resume()` (PRD-HWONB-002, narrowed by PRD-HWONB-015 to a purely
local get-or-create — see its docstring) and the Documents/UMA + Review-
Submit-Transmit-Status functions (PRD-HWONB-012/013 — upload_document,
list_documents, delete_document, get_uma, transmit, refresh_status,
restart_after_rejection, provision_transit) are implemented fully here.
`_save_application_setup_section` (PRD-HWONB-015) is the section that now
actually calls GP's Create Application.
`_save_business_section`/`_save_processing_section` (PRD-HWONB-003/004),
`_save_addresses` (PRD-HWONB-005), and `_save_accounts`/
`list_accounts_with_sync` (PRD-HWONB-006) are also implemented.
`save_section()`'s dispatcher gains one new elif per section as each
vertical-slice developer's schema module is created.
"""

from __future__ import annotations

import logging
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional

from sqlalchemy.orm import Session

from src.apps.merchant_onboarding import crud
from src.apps.merchant_onboarding.client.base_boarding_client import base_boarding_client
from src.apps.merchant_onboarding.helpers import field_mapping, status_machine
from src.apps.merchant_onboarding.helpers.file_sniff import content_matches_declared_type
from src.apps.merchant_onboarding.models.account import MerchantOnboardingAccount
from src.apps.merchant_onboarding.models.application import MerchantOnboardingApplication
from src.apps.merchant_onboarding.models.document import MerchantOnboardingDocument
from src.apps.merchant_onboarding.models.owner import MerchantOnboardingOwner
from src.apps.merchant_onboarding.schemas.card_types import CardTypesRequest, CardTypesResponse
from src.apps.merchant_onboarding.schemas.owners import OwnerItem
from src.apps.payment_providers import services as provider_services
from src.apps.payment_providers.helpers.credentials import decrypt_credential
from src.core.config import settings
from src.core.exceptions import APIException, BadRequestError, ValidationException
from src.events.base import BaseEvent
from src.events.dispatcher import EventDispatcher

logger = logging.getLogger(__name__)

# Human labels for the N-2 section-sequencing gate's combined error message —
# GP's real error text names both missing sections when more than one
# prerequisite is unmet (PRD-HWONB-001 §13.1 N-2: "...business information
# and processing information"). Used by owners/card_types below — the other
# already-merged sections (business/processing/addresses/accounts/products/
# training_activation) each enforce their own N-2 gate inline instead.
_PREREQUISITE_LABELS = {
    "application_setup": "application setup",
    "business": "business information",
    "processing": "processing information",
    "training_activation": "training & activation",
}

# GP's exact required-section list on a failed transmit (PRD-HWONB-013 §1.2,
# authoritative word-for-word wording) — local key -> the label GP uses, in
# GP's own listed order, so a local pre-check 422 reads identically to a real
# GP 40008 response. `business` is checked separately first (see transmit())
# since GP's own error text never names it — attachments/every other section
# already can't be pushed without it (N-2), so it's belt-and-suspenders here.
_TRANSMIT_REQUIRED_SECTIONS: List[tuple[str, str]] = [
    ("processing", "ProcessingInformation"),
    ("accounts", "BankAccount"),
    ("card_types", "CardTypes"),
    ("addresses", "Addresses"),
    ("owners", "Owners"),
    ("products", "Products"),
    ("training_activation", "Services"),
    ("fees", "Fee Schedule"),
    ("documents", "Docs & Signing"),
]


def _section_pushed(section_state: Optional[Dict[str, Any]], key: str) -> bool:
    entry = (section_state or {}).get(key) or {}
    return bool(entry.get("pushed"))


def _utcnow_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


# Partial next-step chain — used by `_finalize_section_save` (live GP success)
# AND `_bootstrap_not_ready_result` (disabled-mode graceful degradation, see
# its docstring/fix note) for application_setup/products/training_activation/fees.
# The other section developers (PRD-HWONB-003..008) advance `current_step`
# inline in their own `_save_*`/disabled-mode branch instead of via this
# table; do not remove entries other than your own if you add to this dict.
# Full chain per PRD-HWONB-001 §6.1 (PRD-HWONB-015 inserts application_setup
# BEFORE business and retires PRD-HWONB-014's pricing_tier step — still 12
# steps, association/leadSource/salesRep are now all collected upfront):
# application_setup -> business -> processing -> addresses -> accounts ->
# owners -> card_types -> products -> training_activation -> fees ->
# documents -> review.
# "fees" -> "documents" is included even though `_save_fees_section`'s own
# live-success branch advances current_step inline rather than through this
# table, so that `_bootstrap_not_ready_result`'s disabled-mode path for fees
# advances current_step the same way its live-success path does.
_NEXT_STEP: Dict[str, str] = {
    "application_setup": "business",
    "products": "training_activation",
    "training_activation": "fees",
    "fees": "documents",
}


async def start_or_resume(db: Session, merchant) -> MerchantOnboardingApplication:
    """
    PRD-HWONB-001 §7.3 — get-or-create the application row and return the
    full resumable row. This backs `GET /onboarding` and is called at the
    top of every other onboarding endpoint to resolve `application`.

    PRD-HWONB-015: this used to also transparently create the GP
    application on first call (PRD-HWONB-002's "invisible bootstrap"). That
    responsibility has moved to `_save_application_setup_section()` — GP's
    Create Application call requires the merchant's own `association`/
    `leadSource`/`salesRep` choices now, so it can no longer fire on a bare
    `GET /onboarding` before the merchant has picked anything. This function
    is purely local: get-or-create the Postgres row, never call GP. A new
    row starts with `tsys_app_id=None` and `current_step="application_setup"`
    (`crud.create_application_row`) — the wizard's existing
    `current_step`-driven routing naturally lands a new merchant on the
    Application Setup screen first, with no special-casing needed here.
    """
    application = crud.get_active_application(db, merchant.id)
    if application is None:
        application = crud.create_application_row(db, merchant.id)
        db.commit()

    return _attach_counts(db, application)


def _attach_counts(db: Session, application: MerchantOnboardingApplication) -> MerchantOnboardingApplication:
    """
    Populate transient (non-column) resume-summary counts on the ORM
    instance so the router can build OnboardingApplicationResponse without
    a second round-trip. These attributes are NOT mapped columns — they
    only exist on this in-memory instance for the duration of the request.
    """
    application.owners_count = crud.count_owners(db, application.id)  # type: ignore[attr-defined]
    application.accounts_count = crud.count_accounts(db, application.id)  # type: ignore[attr-defined]
    application.documents_count = crud.count_documents(db, application.id)  # type: ignore[attr-defined]
    return application


# ---------------------------------------------------------------------------
# Section-sequencing gate (N-2) — shared helpers for owners/card_types below.
# ---------------------------------------------------------------------------


def _missing_prerequisites(application: MerchantOnboardingApplication, section: str) -> List[str]:
    """Which of `section`'s prerequisite sections (field_mapping.SECTION_SEQUENCING_PREREQUISITES) aren't pushed yet."""
    prereqs = field_mapping.SECTION_SEQUENCING_PREREQUISITES.get(section, [])
    section_state = application.section_state or {}
    return [p for p in prereqs if not (section_state.get(p) or {}).get("pushed")]


def _sequencing_error_message(missing: List[str]) -> str:
    """
    Mirrors GP's real combined error text (N-2): "Please complete business
    information" for a single missing prerequisite, "...business information
    and processing information" when more than one is missing (this is
    exactly card_types' case — it needs both business AND processing).
    """
    labels = [_PREREQUISITE_LABELS.get(m, m) for m in missing]
    return f"Please complete {' and '.join(labels)} before continuing."


def _advance_current_step_if_needed(
    db: Session, application: MerchantOnboardingApplication, from_step: str, to_step: str
) -> None:
    if application.current_step == from_step:
        crud.update_application(db, application, current_step=to_step)


# ---------------------------------------------------------------------------
# Owners & Principals — PRD-HWONB-007
# ---------------------------------------------------------------------------


def _owner_to_gp_payload(row: MerchantOnboardingOwner) -> Dict[str, Any]:
    """
    Build the GP `POST /applications/{appId}/owners` request body for one
    owner row — camelCase, only the fields relevant to whichever branch
    (`nonUsCitizen`) this owner is in. Decrypts `ssn_enc` just before
    sending (never logged, never returned to the API — PRD-HWONB-001 AC-13).
    """
    payload: Dict[str, Any] = {
        "firstName": row.first_name,
        "lastName": row.last_name,
        "contactTitle": row.contact_title,
        "phoneNumber": row.phone_number,
        "phoneCountryCode": row.phone_country_code,
        "email": row.email,
        "dob": row.dob.isoformat() if row.dob else None,
        "ownerPercent": row.owner_percent,
        "nonUsCitizen": row.non_us_citizen,
        "appSigner": row.app_signer,
        "personalGuarantor": row.personal_guarantor,
        "beneficialOwner": row.beneficial_owner,
        "individualWithControl": row.individual_with_control,
    }

    if row.non_us_citizen:
        payload.update(
            {
                "passportCountry": row.passport_country,
                "passportNumber": row.passport_number,
                "addressLine1": row.address_line1,
                "addressLine2": row.address_line2,
                "foreignCity": row.foreign_city,
                "foreignState": row.foreign_state,
                "foreignZipCode": row.foreign_zip_code,
                "foreignCountry": row.foreign_country,
            }
        )
    else:
        if row.ssn_enc:
            payload["ssn"] = decrypt_credential(row.ssn_enc)
        payload["creditPullWaive"] = bool(row.credit_pull_waive)
        payload.update(
            {
                "streetNumber": row.street_number,
                "streetName": row.street_name,
                "streetDirection": row.street_direction,
                "streetType": row.street_type,
                "unitNumber": row.unit_number,
                "city": row.city,
                "state": row.state,
                "zipCode": row.zip_code,
                "country": row.country,
            }
        )

    return {k: v for k, v in payload.items() if v is not None}


async def save_owners(
    db: Session,
    application: MerchantOnboardingApplication,
    owners: List[OwnerItem],
) -> List[MerchantOnboardingOwner]:
    """
    `PUT /onboarding/owners` — PRD-HWONB-007 §4.

    1. N-2 gate: owners requires `business` pushed first.
    2. Persist the whole owner list to Postgres BEFORE any GP call (resume
       invariant, PRD-HWONB-001 §7.2) — crud.upsert_owners() is the
       authoritative cap-enforcement layer (<=5 owners, role-count caps,
       exactly-1-individualWithControl, ownerPercent sums to 100,
       ownershipType-dependent SOLE/CNP forcing + ssn requirement).
    3. Push each owner individually (`POST .../owners` × N, PRD-007 §4A —
       create has no `{ownerId}`, GP generates and returns it). One owner's
       GP rejection does not block the others from being pushed.
    4. Update `section_state["owners"]` — `pushed=True` only if every owner
       pushed cleanly; per-row `tsys_errors` always available on each
       returned row for the wizard (AC-07 pattern).
    """
    missing = _missing_prerequisites(application, "owners")
    if missing:
        raise BadRequestError(message=_sequencing_error_message(missing))

    ownership_type = (application.business_data or {}).get("ownershipType")

    # Snapshot each existing row's GP payload BEFORE the upsert mutates it —
    # crud.upsert_owners() updates rows in place below, so this is the only
    # way to later tell "resubmitted the same owner data again" apart from
    # "changed it" (used by the GP-push skip below, mirrors the
    # prior_ciphertext snapshot in _save_accounts). Decrypt failures fall
    # open (row left out of the map, so the skip below never fires for it
    # and it always gets a real push) rather than blocking the whole
    # request over one unrelated row's bad ciphertext.
    prior_gp_payloads: Dict[int, Dict[str, Any]] = {}
    for row in crud.list_owners(db, application.id):
        try:
            prior_gp_payloads[row.id] = _owner_to_gp_payload(row)
        except Exception:
            logger.warning(
                "save_owners: prior-payload snapshot failed for owner_id=%s — "
                "skip check disabled for this row, will always push",
                row.id,
            )

    # snake_case dicts, `id` included (needed by crud.upsert_owners() to
    # route each owner to an update-existing-row vs. create-new-row path).
    owners_dicts = [o.model_dump(mode="json") for o in owners]

    try:
        rows = crud.upsert_owners(db, application.id, owners_dicts, ownership_type=ownership_type)
    except crud.OwnerCapViolation as exc:
        db.rollback()
        raise ValidationException(message=str(exc)) from exc

    # Resume invariant — Postgres write lands before any GP call.
    db.commit()

    if not settings.TSYS_BOARDING_ENABLED or application.tsys_app_id is None:
        # Onboarding not live for this deployment yet / bootstrap hasn't run —
        # rows are saved locally. `pushed` MUST be True here (see
        # `_save_business_section`'s comment) — previously this branch never
        # touched `section_state["owners"]` at all, which (combined with the
        # business-side bug) left the wizard permanently stuck on Owners too.
        logger.warning(
            "save_owners: TSYS_BOARDING_ENABLED=%s tsys_app_id=%s — "
            "rows saved locally, GP push skipped (application_id=%s)",
            settings.TSYS_BOARDING_ENABLED,
            application.tsys_app_id,
            application.id,
        )
        section_state = dict(application.section_state or {})
        section_state["owners"] = {
            "pushed": True,
            "tsys_errors": [
                {
                    "field": "_system",
                    "message": "Global Payments boarding is not enabled in this environment.",
                }
            ],
            "pushed_at": None,
        }
        update_fields: Dict[str, Any] = {"section_state": section_state}
        if application.current_step == "owners":
            update_fields["current_step"] = "card_types"
        crud.update_application(db, application, **update_fields)
        db.commit()
        return rows

    any_errors = False
    all_tsys_errors: List[Dict[str, Any]] = []
    for row in rows:
        gp_payload = _owner_to_gp_payload(row)

        # Skip re-pushing to GP when this owner was already synced
        # successfully before and nothing GP-relevant changed on this
        # resubmission. GP's owner endpoint is create-only (PRD-007 §4A —
        # no `{ownerId}` update route exists), so blindly re-pushing an
        # unchanged owner creates a SECOND owner record on GP's side for
        # the same person. That doesn't get rejected up front the way a
        # duplicate account number does — it silently succeeds — but once
        # two GP-side owners end up flagged individualWithControl=true (the
        # original create + the duplicate), GP starts rejecting further
        # pushes with "individualControlCount must not exceed 1", which
        # can never be fixed by resubmitting again.
        #
        # Gated on `row.tsys_synced_at is not None` (set only on a prior
        # successful push, never on failure) rather than mirroring
        # _save_accounts' unconditional-on-match skip: a row that has never
        # actually reached GP yet (e.g. saved once while
        # TSYS_BOARDING_ENABLED was off) must still get its first real push
        # even if its data hasn't changed since that local-only save.
        prior_payload = prior_gp_payloads.get(row.id)
        if prior_payload is not None and row.tsys_synced_at is not None:
            try:
                unchanged = gp_payload == prior_payload
            except Exception:
                logger.warning(
                    "save_owners: comparison failed for owner_id=%s — "
                    "falling open to a real GP push",
                    row.id,
                )
                unchanged = False
            if unchanged:
                row.tsys_errors = None
                continue

        envelope = await base_boarding_client.push_owner(
            application.tsys_app_id, gp_payload, merchant_id=application.merchant_id, application_id=application.id
        )
        if envelope["ok"]:
            data = envelope["data"] or {}
            owner_ref = data.get("ownerId") or data.get("id") or data.get("ownerID")
            if owner_ref is not None:
                row.tsys_owner_ref = str(owner_ref)
            row.tsys_synced_at = datetime.now(timezone.utc)
            row.tsys_errors = None
        else:
            row.tsys_errors = envelope["tsys_errors"]
            any_errors = True
            all_tsys_errors.extend(envelope["tsys_errors"] or [])

    section_state = dict(application.section_state or {})
    section_state["owners"] = {
        "pushed": not any_errors,
        "tsys_errors": all_tsys_errors or None,
        "pushed_at": datetime.now(timezone.utc).isoformat() if not any_errors else None,
    }
    crud.update_application(db, application, section_state=section_state)
    if not any_errors:
        _advance_current_step_if_needed(db, application, "owners", "card_types")

    db.commit()
    return rows


# ---------------------------------------------------------------------------
# Card Types — PRD-HWONB-008
# ---------------------------------------------------------------------------


async def save_card_types(
    db: Session,
    application: MerchantOnboardingApplication,
    payload: CardTypesRequest,
) -> CardTypesResponse:
    """
    `PUT /onboarding/card-types` — PRD-HWONB-008 §3.

    Single-object section (unlike owners/accounts) — snapshotted onto
    `application.card_types_data`, gated on BOTH `business` AND
    `processing` having been pushed first (N-2's broadest gate; GP's real
    error names both: "...business information and processing information").
    Flat 1:1 mapping, no reshaping (PRD-008 §3/§5) — `model_dump(by_alias=True,
    exclude_none=True)` IS the GP request body.
    """
    missing = _missing_prerequisites(application, "card_types")
    if missing:
        raise BadRequestError(message=_sequencing_error_message(missing))

    # Resume invariant — Postgres write before the GP push. `by_alias=True`
    # keeps the stored snapshot camelCase (matching CardTypesResponse and
    # the wizard's `application.card_types_data.amexRequested`-style reads)
    # instead of the bare snake_case attribute names — without it, a
    # merchant returning to this step after any full reload (re-login,
    # hard refresh) saw a blank form because the keys didn't match.
    crud.update_application(db, application, card_types_data=payload.model_dump(mode="json", by_alias=True))
    db.commit()

    if not settings.TSYS_BOARDING_ENABLED or application.tsys_app_id is None:
        logger.warning(
            "save_card_types: TSYS_BOARDING_ENABLED=%s tsys_app_id=%s — "
            "snapshot saved locally, GP push skipped (application_id=%s)",
            settings.TSYS_BOARDING_ENABLED,
            application.tsys_app_id,
            application.id,
        )
        # `pushed` MUST be True in section_state (see `_save_business_section`'s
        # comment) — previously this branch never touched
        # `section_state["card_types"]` at all. `card_types` is currently the
        # last N-2-gated section (owners/card_types both require business,
        # card_types additionally requires processing), so this closes the
        # last gap in the sequencing-trap chain.
        degraded_tsys_errors = [
            {
                "field": "_system",
                "message": "Global Payments boarding is not enabled in this environment.",
            }
        ]
        section_state = dict(application.section_state or {})
        section_state["card_types"] = {
            "pushed": True,
            "tsys_errors": degraded_tsys_errors,
            "pushed_at": None,
        }
        update_fields: Dict[str, Any] = {"section_state": section_state}
        if application.current_step == "card_types":
            update_fields["current_step"] = "products"
        crud.update_application(db, application, **update_fields)
        db.commit()
        return CardTypesResponse(**payload.model_dump(), pushed=True, tsys_errors=degraded_tsys_errors)

    gp_payload = payload.model_dump(by_alias=True, exclude_none=True)
    envelope = await base_boarding_client.push_card_types(
        application.tsys_app_id, gp_payload, merchant_id=application.merchant_id, application_id=application.id
    )

    section_state = dict(application.section_state or {})
    if envelope["ok"]:
        section_state["card_types"] = {
            "pushed": True,
            "tsys_errors": None,
            "pushed_at": datetime.now(timezone.utc).isoformat(),
        }
        crud.update_application(db, application, section_state=section_state)
        _advance_current_step_if_needed(db, application, "card_types", "products")
    else:
        section_state["card_types"] = {
            "pushed": False,
            "tsys_errors": envelope["tsys_errors"],
            "pushed_at": None,
        }
        crud.update_application(db, application, section_state=section_state)

    db.commit()

    return CardTypesResponse(
        **payload.model_dump(),
        pushed=envelope["ok"],
        tsys_errors=envelope["tsys_errors"],
    )


# ---------------------------------------------------------------------------
# Stubs for the vertical-slice developers
# ---------------------------------------------------------------------------


async def save_section(
    db: Session,
    application: MerchantOnboardingApplication,
    section: str,
    payload: Dict[str, Any],
    *,
    named_misc_fee_overrides: Optional[Dict[str, float]] = None,
) -> Dict[str, Any]:
    """
    Generic per-section save dispatcher — PUT /onboarding/{section}.

    Intended shape once implemented (do not deviate without updating every
    step PRD's router wiring):
      1. Validate `section` against the known section names (business,
         processing, addresses, accounts, owners, card_types, products,
         training_activation, fees) and check
         helpers.field_mapping.SECTION_SEQUENCING_PREREQUISITES before
         pushing (N-2 — GP rejects out-of-order section pushes).
      2. Persist the local snapshot (`{section}_data` column /
         merchant_onboarding_owners / merchant_onboarding_accounts row) to
         Postgres BEFORE calling GP (PRD-HWONB-001 §7.2 — a browser close
         mid-request must never lose more than "this section needs
         re-saving").
      3. Call the matching `base_boarding_client.push_*()` method.
      4. On success: update `section_state[section] = {"pushed": True,
         "tsys_errors": None, "pushed_at": now}`, advance `current_step` if
         this was the current step, commit.
      5. On GP tsys_errors: update `section_state[section]` with the errors
         but do NOT advance current_step; return them so the router can
         surface field-level errors to the wizard (AC-07).

    Each of the 9 sections has its own PRD (003-011) with the exact request
    shape — this function is filled in incrementally as each section's
    schema module (schemas/business.py, schemas/processing.py, ...) is
    created by that section's developer. Until then, no section schemas
    exist, so this dispatcher has nothing to validate against.

    `business`, `processing`, `addresses`, `accounts`, `products`,
    `training_activation`, `owners`, and `card_types` are implemented
    (PRD-HWONB-003/004/005/006/009/010/007/008). `owners`/`card_types`
    parse the raw dict into their Pydantic schema here and delegate to
    save_owners()/save_card_types() — the router itself calls those two
    directly with an already-validated request model rather than through
    this dict-based dispatcher (each needs a different request shape: a
    list for owners, a flat object for card_types), so this branch mainly
    exists for any other caller that only has a raw dict + section name.
    Every other section still falls through to NotImplementedError until its
    own developer adds a branch here — do not restructure this dispatch
    shape, other developers are wiring their sections in parallel.
    """
    if section == "application_setup":
        return await _save_application_setup_section(db, application, payload)
    elif section == "business":
        return await _save_business_section(db, application, payload)
    elif section == "processing":
        return await _save_processing_section(db, application, payload)
    elif section == "addresses":
        return await _save_addresses(db, application, payload)
    elif section == "accounts":
        return await _save_accounts(db, application, payload)
    elif section == "products":
        return await _save_products_section(db, application, payload)
    elif section == "training_activation":
        return await _save_training_activation_section(db, application, payload)
    elif section == "owners":
        owners = [OwnerItem.model_validate(o) for o in payload.get("owners", [])]
        rows = await save_owners(db, application, owners)
        return {
            "owners": rows,
            "any_errors": any(bool(r.tsys_errors) for r in rows),
        }
    elif section == "card_types":
        card_types_request = CardTypesRequest.model_validate(payload)
        response = await save_card_types(db, application, card_types_request)
        return response.model_dump()
    elif section == "fees":
        return await _save_fees_section(db, application, payload, named_misc_fee_overrides)

    raise NotImplementedError(
        f"save_section({section!r}) is implemented per-section by the PRD-HWONB-003..011 developers "
        "as each section's schema module is created."
    )


async def _save_business_section(
    db: Session,
    application: MerchantOnboardingApplication,
    payload: Dict[str, Any],
) -> Dict[str, Any]:
    """
    PRD-HWONB-003 §3 — Business Information save.

    `payload` is already-validated `BusinessInformationRequest.to_gp_payload()`
    (router validates via the Pydantic request model before calling here).

    Resume invariant (PRD-HWONB-001 §7.2 / §9.6 convention #1): the local
    snapshot is written to Postgres BEFORE the GP push — a browser close
    mid-request never loses more than "this section needs re-saving".

    PRD-HWONB-015 — deliberately has NO `application_setup` N-2 gate (unlike
    every prerequisite-checked section below): normal wizard navigation
    can't reach Business before Application Setup (it's the new Step 1)
    regardless, and adding one here would require every existing test/tool
    that pushes `business` directly (bypassing the wizard's own routing) to
    first push `application_setup` too, for no real behavioral benefit.
    """
    crud.update_application(db, application, business_data=payload)
    db.commit()

    section_state: Dict[str, Any] = dict(application.section_state or {})

    if not settings.TSYS_BOARDING_ENABLED or application.tsys_app_id is None:
        # Onboarding not live for this deployment (or the invisible bootstrap
        # hasn't minted a GP application yet) — the local snapshot above is
        # still saved, but there's no GP call to make. Fail gracefully rather
        # than crash, matching PRD-HWONB-002's "onboarding not yet available"
        # posture for this exact condition.
        #
        # `pushed` MUST be True here, not False — this boolean is what every
        # downstream section's N-2 sequencing gate checks
        # (_missing_prerequisites / `section_state["business"]["pushed"]`).
        # If it stayed False, Business could never satisfy its own
        # prerequisite in disabled mode and the entire wizard would be
        # permanently stuck after step 1 in every dev/test/CI environment
        # (the real TSYS_BOARDING_ENABLED=False default) — this was a real
        # bug, see TestSequencingTrapUnderDisabledMode in
        # tests/test_merchant_onboarding/test_section_saves.py. The
        # `tsys_errors` informational note is kept (non-blocking) so the
        # wizard can still show a "not pushed to Global Payments yet" banner
        # if it wants to.
        section_state["business"] = {
            "pushed": True,
            "tsys_errors": [
                {
                    "field": "_system",
                    "message": "Global Payments boarding is not enabled in this environment.",
                }
            ],
            "pushed_at": None,
        }
        update_fields: Dict[str, Any] = {"section_state": section_state}
        if application.current_step == "business":
            update_fields["current_step"] = "processing"
        crud.update_application(db, application, **update_fields)
        db.commit()
        return {
            "saved": True,
            "tsys_errors": section_state["business"]["tsys_errors"],
            "section_state": section_state,
            "status": application.status,
        }

    envelope = await base_boarding_client.push_business(
        application.tsys_app_id, payload, merchant_id=application.merchant_id, application_id=application.id
    )

    if envelope["ok"]:
        section_state["business"] = {
            "pushed": True,
            "tsys_errors": None,
            "pushed_at": _utcnow_iso(),
        }
        update_fields: Dict[str, Any] = {"section_state": section_state}
        if application.current_step == "business":
            update_fields["current_step"] = "processing"
        crud.update_application(db, application, **update_fields)
    else:
        logger.info(
            "_save_business_section: GP push_business failed for application_id=%s tsys_errors=%s",
            application.id,
            envelope["tsys_errors"],
        )
        section_state["business"] = {
            "pushed": False,
            "tsys_errors": envelope["tsys_errors"],
            "pushed_at": None,
        }
        crud.update_application(db, application, section_state=section_state)

    db.commit()

    return {
        "saved": envelope["ok"],
        "tsys_errors": envelope["tsys_errors"],
        "section_state": section_state,
        "status": application.status,
    }


async def _save_processing_section(
    db: Session,
    application: MerchantOnboardingApplication,
    payload: Dict[str, Any],
) -> Dict[str, Any]:
    """
    PRD-HWONB-004 §3 — Processing Information save.

    N-2 sequencing gate (PRD-HWONB-001 §13.1): GP itself rejects
    `processingInformation` if `business` hasn't been pushed yet — surface
    that as a clean local 422 ("Please complete business information",
    matching GP's own real error wording) instead of letting a raw GP 400
    reach the wizard.

    Also enforces the cross-section conditional from PRD-HWONB-004 §1.2 that
    schemas/processing.py cannot validate on its own (it needs
    `businessDetails.ownershipType` from the already-saved Business section):
    `governmentIdType` is required when ownershipType == "GE";
    `businessIdType` is required when ownershipType != "GE".
    """
    section_state: Dict[str, Any] = dict(application.section_state or {})
    business_state = section_state.get("business") or {}
    if not business_state.get("pushed"):
        raise APIException(
            status_code=422,
            message="Please complete business information",
            error=[
                {
                    "field": "_prerequisite",
                    "message": "Please complete business information",
                }
            ],
        )

    business_data = application.business_data or {}
    ownership_type = (business_data.get("businessDetails") or {}).get("ownershipType")
    patriot_act = payload.get("applicationPatriotAct") or {}
    cross_field_errors = []
    if ownership_type == "GE" and not patriot_act.get("governmentIdType"):
        cross_field_errors.append(
            {
                "field": "applicationPatriotAct.governmentIdType",
                "message": "governmentIdType is required when the business's ownershipType is GE",
            }
        )
    if ownership_type is not None and ownership_type != "GE" and not patriot_act.get("businessIdType"):
        cross_field_errors.append(
            {
                "field": "applicationPatriotAct.businessIdType",
                "message": "businessIdType is required when the business's ownershipType is not GE",
            }
        )
    if cross_field_errors:
        raise APIException(
            status_code=422,
            message="Processing information is incomplete for this business's ownership type.",
            error=cross_field_errors,
        )

    # Resume invariant — persist local snapshot before the GP push.
    crud.update_application(db, application, processing_data=payload)
    db.commit()

    if not settings.TSYS_BOARDING_ENABLED or application.tsys_app_id is None:
        # `pushed` MUST be True — see the identical comment in
        # `_save_business_section` above. Without this, `card_types`'s N-2
        # gate (which requires BOTH business and processing pushed) could
        # never be satisfied in disabled mode either.
        section_state["processing"] = {
            "pushed": True,
            "tsys_errors": [
                {
                    "field": "_system",
                    "message": "Global Payments boarding is not enabled in this environment.",
                }
            ],
            "pushed_at": None,
        }
        update_fields: Dict[str, Any] = {"section_state": section_state}
        if application.current_step == "processing":
            update_fields["current_step"] = "addresses"
        crud.update_application(db, application, **update_fields)
        db.commit()
        return {
            "saved": True,
            "tsys_errors": section_state["processing"]["tsys_errors"],
            "section_state": section_state,
            "status": application.status,
        }

    envelope = await base_boarding_client.push_processing(
        application.tsys_app_id, payload, merchant_id=application.merchant_id, application_id=application.id
    )

    if envelope["ok"]:
        section_state["processing"] = {
            "pushed": True,
            "tsys_errors": None,
            "pushed_at": _utcnow_iso(),
        }
        update_fields: Dict[str, Any] = {"section_state": section_state}
        if application.current_step == "processing":
            update_fields["current_step"] = "addresses"
        crud.update_application(db, application, **update_fields)
    else:
        logger.info(
            "_save_processing_section: GP push_processing failed for application_id=%s tsys_errors=%s",
            application.id,
            envelope["tsys_errors"],
        )
        section_state["processing"] = {
            "pushed": False,
            "tsys_errors": envelope["tsys_errors"],
            "pushed_at": None,
        }
        crud.update_application(db, application, section_state=section_state)

    db.commit()

    return {
        "saved": envelope["ok"],
        "tsys_errors": envelope["tsys_errors"],
        "section_state": section_state,
        "status": application.status,
    }


# ---------------------------------------------------------------------------
# PRD-HWONB-005 — Addresses (simple single-object section, no dedicated table)
# ---------------------------------------------------------------------------


def _build_gp_addresses_payload(payload: Dict[str, Any]) -> Dict[str, Any]:
    """
    Reshape HubWallet's validated `AddressesRequest` dict into GP's exact
    keyed-object envelope (C-11): `{dba, legal, mailing, chargeback}`. When a
    conditional sub-object's `isSameAsDba` is true, every other field on it
    is omitted from the GP payload entirely (§3/AC-2) even if the client
    happened to send stale values — GP only ever sees `{"isSameAsDba": true}`
    for that key in that case.
    """
    gp_payload: Dict[str, Any] = {"dba": payload["dba"]}
    for key in field_mapping.ADDRESS_CONDITIONAL_KEYS:
        sub = payload.get(key) or {}
        if sub.get("isSameAsDba"):
            gp_payload[key] = {"isSameAsDba": True}
        else:
            gp_payload[key] = {k: v for k, v in sub.items() if v is not None}
    return gp_payload


async def _save_addresses(
    db: Session, application: MerchantOnboardingApplication, payload: Dict[str, Any]
) -> Dict[str, Any]:
    """
    PUT /onboarding/addresses — PRD-HWONB-005.

    `addresses` requires `business` to have been pushed first (N-2 /
    field_mapping.SECTION_SEQUENCING_PREREQUISITES) — checked here so an
    out-of-order save gets a clean local error instead of a raw GP 400.
    """
    section_state = dict(application.section_state or {})
    missing_prereqs = _missing_prerequisites(application, "addresses")
    if missing_prereqs:
        # Standardized to the same 422 APIException contract as
        # processing/owners/card_types for the identical "business not yet
        # pushed" condition (previously this returned a 200 with an inline
        # `tsys_errors` entry instead — an inconsistent HTTP contract for the
        # frontend to special-case for no good reason).
        raise APIException(
            status_code=422,
            message=_sequencing_error_message(missing_prereqs),
            error=[
                {
                    "field": "_prerequisite",
                    "message": _sequencing_error_message(missing_prereqs),
                }
            ],
        )

    # 1. Persist the local snapshot to Postgres BEFORE calling GP (resume
    #    invariant, PRD-HWONB-001 §7.2) — a browser close mid-request loses
    #    nothing worse than "addresses needs re-saving".
    crud.update_application(db, application, addresses_data=payload)
    db.commit()

    if not settings.TSYS_BOARDING_ENABLED or application.tsys_app_id is None:
        # Onboarding not live for this deployment yet / bootstrap hasn't run —
        # local snapshot is saved. `pushed` MUST be True (see
        # `_save_business_section`'s comment) so any later section that ever
        # gains an `addresses` prerequisite (and, more importantly, so a
        # `GET /onboarding` banner keyed off section_state) doesn't read this
        # as "never touched" the way it did before this fix.
        logger.warning(
            "_save_addresses: TSYS_BOARDING_ENABLED=%s tsys_app_id=%s — "
            "snapshot saved locally, GP push skipped (application_id=%s)",
            settings.TSYS_BOARDING_ENABLED,
            application.tsys_app_id,
            application.id,
        )
        section_state["addresses"] = {
            "pushed": True,
            "tsys_errors": [
                {
                    "field": "_system",
                    "message": "Global Payments boarding is not enabled in this environment.",
                }
            ],
            "pushed_at": None,
        }
        update_fields: Dict[str, Any] = {"section_state": section_state}
        if application.current_step == "addresses":
            update_fields["current_step"] = "accounts"
        crud.update_application(db, application, **update_fields)
        db.commit()
        return {**payload, "tsys_errors": section_state["addresses"]["tsys_errors"]}

    gp_payload = _build_gp_addresses_payload(payload)
    envelope = await base_boarding_client.push_addresses(
        application.tsys_app_id, gp_payload, merchant_id=application.merchant_id, application_id=application.id
    )

    if envelope["ok"]:
        section_state["addresses"] = {
            "pushed": True,
            "tsys_errors": None,
            "pushed_at": datetime.now(timezone.utc).isoformat(),
        }
        # Unlike business/processing/owners/card_types, this section never
        # advanced current_step past "addresses" — a merchant who resumed
        # after this step (e.g. after a session expiry) was redirected back
        # to Addresses regardless of how much further they'd actually
        # completed (sidebar still correctly showed later steps as done).
        update_fields: Dict[str, Any] = {"section_state": section_state}
        if application.current_step == "addresses":
            update_fields["current_step"] = "accounts"
        crud.update_application(db, application, **update_fields)
        db.commit()
        return {**payload, "tsys_errors": None}

    section_state["addresses"] = {
        "pushed": False,
        "tsys_errors": envelope["tsys_errors"],
        "pushed_at": None,
    }
    crud.update_application(db, application, section_state=section_state)
    db.commit()
    return {**payload, "tsys_errors": envelope["tsys_errors"]}


# ---------------------------------------------------------------------------
# PRD-HWONB-006 — Bank Accounts (list-upsert, dedicated table, per-row GP push)
# ---------------------------------------------------------------------------


def _account_row_to_dict(row: MerchantOnboardingAccount) -> Dict[str, Any]:
    """
    Mask sensitive fields for the API response — only the last 4 digits of
    routing/account numbers are ever echoed back (judgment call: not
    specified by the PRD, but there is no legitimate reason for the response
    to round-trip full bank account numbers once saved).
    """
    from src.apps.payment_providers.helpers.credentials import decrypt_credential

    def _last4(enc_value: str) -> Optional[str]:
        try:
            return decrypt_credential(enc_value)[-4:]
        except Exception:
            logger.warning(
                "_account_row_to_dict: failed to decrypt a stored value for account_id=%s",
                row.id,
            )
            return None

    return {
        "id": row.id,
        "routingNumberLast4": _last4(row.routing_number_enc),
        "accountNumberLast4": _last4(row.account_number_enc),
        "accountType": row.account_type,
        "usageTypes": row.usage_types or [],
        "defaultAccount": row.is_default,
        "tsysAccountRef": row.tsys_account_ref,
        "tsysErrors": row.tsys_errors,
    }


def _build_gp_account_payload(item: Dict[str, Any], row: MerchantOnboardingAccount) -> Dict[str, Any]:
    """
    Build one account's GP create body — plaintext routing/account numbers,
    never encrypted values. `item` may omit routingNumber/accountNumber
    (blank means "keep what's on file" — crud.upsert_accounts already left
    `row`'s encrypted columns untouched in that case), so fall back to
    decrypting the row's stored ciphertext for whichever one is missing.
    """
    routing_number = item.get("routingNumber") or decrypt_credential(row.routing_number_enc)
    account_number = item.get("accountNumber") or decrypt_credential(row.account_number_enc)
    gp_payload: Dict[str, Any] = {
        "routingNumber": routing_number,
        "accountNumber": account_number,
        "accountType": item["accountType"],
    }
    if item.get("usageTypes"):
        gp_payload["usageTypes"] = item["usageTypes"]
    # Echo the row's resolved default flag (post crud.upsert_accounts'
    # "first account defaults to true if omitted" promotion), not the raw
    # submitted value, so GP and the local row never disagree.
    gp_payload["defaultAccount"] = bool(row.is_default)
    return gp_payload


async def _save_accounts(
    db: Session, application: MerchantOnboardingApplication, payload: Dict[str, Any]
) -> Dict[str, Any]:
    """
    PUT /onboarding/accounts — PRD-HWONB-006, list-upsert.

    `accounts` has NO sequencing prerequisite (N-2 confirms it is
    independent of the business-first gate — it can be pushed before
    Business). GP's create is per-account, not batch (§4/§4A), so each row
    is pushed individually here, in submission order — not one combined call.
    """
    accounts_payload: List[Dict[str, Any]] = payload.get("accounts", [])

    # Snapshot pre-upsert ciphertext for existing rows — crud.upsert_accounts
    # overwrites routing_number_enc/account_number_enc in place below, so this
    # is the only way to later tell "resubmitted the same numbers again"
    # apart from "changed them" (used by the GP-push skip below).
    prior_ciphertext: Dict[int, Any] = {
        row.id: (row.routing_number_enc, row.account_number_enc)
        for row in crud.list_accounts(db, application.id)
    }

    # 1. Local write FIRST (resume invariant) — cap enforcement (<=3, C-23)
    #    and soft-delete-on-removal both happen inside crud.upsert_accounts.
    rows = crud.upsert_accounts(db, application.id, accounts_payload)
    db.commit()

    if not settings.TSYS_BOARDING_ENABLED or application.tsys_app_id is None:
        logger.warning(
            "_save_accounts: TSYS_BOARDING_ENABLED=%s tsys_app_id=%s — "
            "rows saved locally, GP push skipped (application_id=%s)",
            settings.TSYS_BOARDING_ENABLED,
            application.tsys_app_id,
            application.id,
        )
        # Fix: every other section's disabled-mode branch writes its own
        # `section_state[section]` entry; accounts previously never did,
        # unlike every other section (harmless for the transmit gate itself,
        # which checks `crud.count_accounts` rather than section_state, but
        # inconsistent for any UI/audit logic keyed off section_state).
        # `pushed=True` also matches the "unblock sequencing" fix applied to
        # every other section above (accounts has no N-2 prerequisite of its
        # own, but nothing should ever read a missing entry as "untouched").
        section_state = dict(application.section_state or {})
        section_state["accounts"] = {
            "pushed": True,
            "tsys_errors": [
                {
                    "field": "_system",
                    "message": "Global Payments boarding is not enabled in this environment.",
                }
            ],
            "pushed_at": None,
        }
        crud.update_application(db, application, section_state=section_state)
        _advance_current_step_if_needed(db, application, "accounts", "owners")
        db.commit()
        # NOTE: `AccountsResponse.tsys_errors` is typed as a per-account-index
        # keyed dict (`Optional[Dict[str, List[Dict[str, str]]]]`), not a flat
        # list — the informational `_system` note above isn't tied to any one
        # account row, so it belongs in `section_state` (free-form, not
        # response-schema-validated) only; the response itself stays None
        # here exactly as it did before this fix.
        return {"accounts": [_account_row_to_dict(r) for r in rows], "tsys_errors": None}

    aggregated_errors: Dict[str, Any] = {}
    for idx, row in enumerate(rows):
        item = accounts_payload[idx]

        # Skip re-pushing to GP when nothing about this row's bank numbers
        # actually changed (either resubmitted blank — crud.upsert_accounts
        # left the ciphertext untouched — or retyped identically). This is
        # what stops an unavoidable resubmission (numbers are never echoed
        # back in full, so the merchant can't tell they're "the same") from
        # hitting GP's "Duplicate account number. Account already exists."
        # rejection.
        #
        # Deliberately NOT gated on row.tsys_account_ref/row.tsys_errors:
        # a row can have no ref recorded (GP's create response was lost, or
        # an earlier attempt on this exact data errored) while GP's side
        # already has it — retrying identical data can never resolve that,
        # it just repeats the same rejection forever. Any stale error from
        # a prior submission is cleared here too, since it no longer
        # reflects anything the merchant can act on by resubmitting.
        # (Edge case: a row that was never pushed at all, e.g. because
        # TSYS_BOARDING_ENABLED was off at save time, also gets skipped
        # here if resubmitted unchanged — the merchant would need to
        # retype the numbers to force a genuine push in that scenario.)
        prior = prior_ciphertext.get(row.id)
        if prior:
            routing_in = item.get("routingNumber")
            account_in = item.get("accountNumber")
            try:
                unchanged = (
                    not routing_in or decrypt_credential(prior[0]) == routing_in
                ) and (not account_in or decrypt_credential(prior[1]) == account_in)
            except Exception:
                logger.warning(
                    "_save_accounts: decrypt failed comparing prior value for "
                    "account_id=%s — falling open to a real GP push",
                    row.id,
                )
                unchanged = False
            if unchanged:
                row.tsys_errors = None
                continue

        gp_payload = _build_gp_account_payload(item, row)
        envelope = await base_boarding_client.push_account(
            application.tsys_app_id, gp_payload, merchant_id=application.merchant_id, application_id=application.id
        )

        if envelope["ok"]:
            data = envelope["data"] or {}
            tsys_ref = data.get("accountId") or data.get("id")
            if tsys_ref is not None:
                row.tsys_account_ref = str(tsys_ref)
            row.tsys_synced_at = datetime.now(timezone.utc)
            row.tsys_errors = None
        else:
            tsys_errors = envelope["tsys_errors"] or []
            already_exists = any(
                "duplicate" in str(e.get("message", "")).lower()
                and "already exists" in str(e.get("message", "")).lower()
                for e in tsys_errors
            )
            if already_exists:
                # GP's accounts endpoint is create-only (no update, no
                # lookup-by-number) — this response means an earlier push
                # of this exact data DID succeed on GP's side even though
                # this row never recorded a tsys_account_ref (response
                # lost, or a retry of the same data landed here again).
                # Treat as idempotent success rather than surfacing a
                # rejection the merchant has no way to act on (mirrors the
                # "already transmitted" handling in transmit(), below).
                row.tsys_synced_at = datetime.now(timezone.utc)
                row.tsys_errors = None
            else:
                row.tsys_errors = tsys_errors
                aggregated_errors[str(idx)] = tsys_errors

    db.flush()

    section_state = dict(application.section_state or {})
    section_had_errors = bool(aggregated_errors)
    section_state["accounts"] = {
        "pushed": not section_had_errors,
        "tsys_errors": aggregated_errors or None,
        "pushed_at": None if section_had_errors else datetime.now(timezone.utc).isoformat(),
    }
    crud.update_application(db, application, section_state=section_state)
    if not section_had_errors:
        _advance_current_step_if_needed(db, application, "accounts", "owners")
    db.commit()

    return {
        "accounts": [_account_row_to_dict(r) for r in rows],
        "tsys_errors": aggregated_errors or None,
    }


async def list_accounts_with_sync(
    db: Session, application: MerchantOnboardingApplication
) -> Dict[str, Any]:
    """
    GET /onboarding/accounts — PRD-HWONB-006 §4, Show/Sync. Returns
    HubWallet's local rows (masked) plus GP's own stored copy for manual
    drift reconciliation (e.g. a push GP recorded but HubWallet's request
    timed out on). Does NOT auto-merge the two — there is no confirmed rule
    for which side should win a conflict, so both are surfaced and a human
    (or a future explicit "reconcile" action) decides. Not required before
    Continue — background/manual check only, per PRD §4.
    """
    rows = crud.list_accounts(db, application.id)
    result: Dict[str, Any] = {
        "accounts": [_account_row_to_dict(r) for r in rows],
        "tsys_errors": None,
        "gp_accounts": None,
    }

    if settings.TSYS_BOARDING_ENABLED and application.tsys_app_id is not None:
        envelope = await base_boarding_client.get_accounts(application.tsys_app_id, merchant_id=application.merchant_id, application_id=application.id)
        if envelope["ok"]:
            result["gp_accounts"] = envelope["data"]
        else:
            result["tsys_errors"] = {"_sync": envelope["tsys_errors"]}

    return result


# ---------------------------------------------------------------------------
# PRD-HWONB-009 — Products & Equipment / PRD-HWONB-010 — Training & Activation
# ---------------------------------------------------------------------------


def _resolve_standalone_print_size(equipment: Optional[Dict[str, Any]]) -> Optional[str]:
    """
    Pull GP's per-device print-size hint out of a standaloneEquipment catalog
    response. GP's filtered `/product/standalone/equipment` lookup exposes the
    actual accepted `configurations.printSize` value for the selected
    brand/model/industry/application/source combo under
    `equipment.allowedFeature.appPrintSize` — discovered live 2026-07-22 by
    cross-referencing a create-product rejection ("PrintSize must match an
    accepted value: [S]") against the catalog call the frontend cascade had
    already made ~30s earlier for the same combo, which returned
    appPrintSize="S" (the schema's hidden "M" default was only ever correct
    by coincidence). Returns None (caller keeps the schema default) if the
    hint is missing or not one of GP's accepted S/M/L values.
    """
    if not isinstance(equipment, dict):
        return None
    allowed_feature = equipment.get("allowedFeature")
    if not isinstance(allowed_feature, dict):
        return None
    print_size = allowed_feature.get("appPrintSize")
    return print_size if print_size in ("S", "M", "L") else None


async def _save_products_section(
    db: Session,
    application: MerchantOnboardingApplication,
    payload: Dict[str, Any],
) -> Dict[str, Any]:
    """
    PRD-HWONB-009 — Products & Equipment.

    `payload` is the already-validated (and default-/derivation-applied)
    dict from `schemas.products.StandaloneProduct` /
    `IntegratedProduct.model_dump()` (router.py builds this via the
    `ProductsSavePayload` discriminated union before calling here).

    AC-5: for the Integrated path, `ach` is recomputed here from the
    merchant's saved bank accounts (crud.has_ach_usage_type) and OVERWRITTEN
    regardless of what the client sent — this derivation needs a DB read the
    Pydantic schema layer doesn't have access to.
    """
    if payload.get("productType") == "integrated":
        payload["ach"] = crud.has_ach_usage_type(db, application.id)

    # Persist the local snapshot BEFORE calling GP (§7.2 resume guarantee).
    crud.update_application(db, application, products_data=payload)
    db.commit()

    degraded = _bootstrap_not_ready_result(db, application, "products")
    if degraded is not None:
        return degraded

    gp_payload = payload
    if payload.get("productType") == "integrated":
        # CERT trial (2026-07-16): GP's real create-integrated-product call
        # rejects the literal `integrationProductID` key ("Request expects
        # the mandatory parameter integrationSwId") — resolves the
        # write-field-name judgment call flagged in schemas/products.py's
        # module docstring in favor of `integrationSwId`.
        #
        # CERT re-trial (2026-07-28): the earlier "monthlyFee/setupFee
        # rejected outright" conclusion was wrong — both fields were being
        # sent as a quoted string defaulting to "0.00", and GP's real
        # contract (Developer Guide V1.9 pp.188-191) is a JSON *number* in
        # range 0.01-999,999.99 (0.00 is genuinely below the floor, live
        # CERT-confirmed with swId=129 — not just a type issue). Fixed at
        # the schema level (`schemas/products.py`); both fees now flow
        # through unmodified, no pop needed.
        gp_payload = {**payload}
        gp_payload["integrationSwId"] = gp_payload.pop("integrationProductID")
    elif payload.get("productType") == "standalone":
        # Re-fetch the same filtered standalone equipment catalog the
        # frontend cascade already calls, purely to read the real
        # `appPrintSize` hint for this combo — see
        # `_resolve_standalone_print_size`. Non-fatal on any failure: falls
        # back to the schema's default `configurations.printSize`.
        catalog_envelope = await base_boarding_client.get_standalone_equipment_catalog(
            settings.TSYS_BOARDING_PARTNER,
            association=application.tsys_association or settings.TSYS_BOARDING_ASSOCIATION,
            brand=payload.get("brand"),
            model=payload.get("model"),
            industry=payload.get("industry"),
            application=payload.get("application"),
            source=payload.get("source"),
            merchant_id=application.merchant_id,
            application_id=application.id,
        )
        equipment = None
        if catalog_envelope["ok"] and isinstance(catalog_envelope["data"], dict):
            equipment = catalog_envelope["data"].get("Equipment") or catalog_envelope["data"].get("equipment")
        print_size = _resolve_standalone_print_size(equipment)
        if print_size:
            gp_payload = {**payload, "configurations": {**payload["configurations"], "printSize": print_size}}

    envelope = await base_boarding_client.push_product(
        application.tsys_app_id, gp_payload, merchant_id=application.merchant_id, application_id=application.id
    )
    return _finalize_section_save(db, application, "products", payload, envelope)


async def _save_training_activation_section(
    db: Session,
    application: MerchantOnboardingApplication,
    payload: Dict[str, Any],
) -> Dict[str, Any]:
    """
    PRD-HWONB-010 — Training & Activation.

    `payload` is the GP write body built by
    `schemas.training_activation.TrainingActivationRequest.to_gp_payload()`
    (flat object, no reshaping — PRD §5).
    """
    crud.update_application(db, application, training_activation_data=payload)
    db.commit()

    degraded = _bootstrap_not_ready_result(db, application, "training_activation")
    if degraded is not None:
        return degraded

    envelope = await base_boarding_client.push_training_activation(
        application.tsys_app_id, payload, merchant_id=application.merchant_id, application_id=application.id
    )
    return _finalize_section_save(db, application, "training_activation", payload, envelope)


# ---------------------------------------------------------------------------
# PRD-HWONB-015 — Application Setup (association / leadSource / salesRep)
# ---------------------------------------------------------------------------
#
# Supersedes PRD-HWONB-002's "invisible bootstrap" (§1: "not a screen the
# merchant sees") and PRD-HWONB-014's bootstrap-then-PATCH flow for
# `association` alone. All three of GP's Create Application merchant-facing
# values — association, leadSource, salesRep — are now collected together
# in a dedicated Application Setup screen that runs BEFORE Business
# Information (Step 1 of 12), and are sent on the real `POST /applications`
# call itself, not patched in afterward. `partner` (fixed "HUBWALLET"),
# `merchantApplicationType` (fixed "NEW_MERCHANT"), and `sBank` (fixed
# "SYN" — the LLC sponsor bank) remain config-driven, never merchant-facing.
async def _save_application_setup_section(
    db: Session,
    application: MerchantOnboardingApplication,
    payload: Dict[str, Any],
) -> Dict[str, Any]:
    """
    `payload` is `schemas.application_setup.ApplicationSetupRequest.model_dump()`
    — `{"association": "<6-digit code>", "leadSource": "<code>", "salesRep": <int>}`.

    Idempotency (mirrors PRD-HWONB-002 AC-2's intent, now scoped to this
    section): GP's Create Application is a create, not an update — if
    `tsys_app_id` is already set, this is a resumed merchant who already has
    a GP application, so this is a no-op success rather than a second
    `POST /applications` call (which would either 400 or silently orphan a
    second appId). A resubmission with different values than what's already
    stored is logged but never fails the request over it — the wizard has
    no legitimate reason to resubmit this screen once it's been pushed.
    """
    if application.tsys_app_id is not None:
        if (
            application.tsys_association != payload["association"]
            or application.tsys_lead_source != payload["leadSource"]
            or application.tsys_sales_rep != payload["salesRep"]
        ):
            logger.warning(
                "_save_application_setup_section: resubmission with different "
                "values for application_id=%s (tsys_app_id=%s already set) — "
                "ignoring, GP's Create Application cannot be re-called.",
                application.id,
                application.tsys_app_id,
            )
        return {
            "section": "application_setup",
            "pushed": True,
            "tsys_errors": None,
            "current_step": application.current_step,
            "data": None,
        }

    # Reject a TransFreedom-only association up front — no GP-documented
    # transFreedomBundle value exists for self-service, so a merchant who
    # picks one here can never complete the Fees step later no matter which
    # plan/option they choose there. Catches associations whose *name*
    # doesn't say TransFreedom too (router.py's lookup-level name filter is
    # cheap defense-in-depth, not sufficient by itself — see
    # `association_is_viable`'s docstring). Checked before persisting
    # anything so a bad choice never gets saved. Skipped when boarding is
    # disabled (dev/test) — same gate `TSYS_BOARDING_ENABLED` guards below —
    # since there's no live GP catalog to check against.
    if settings.TSYS_BOARDING_ENABLED and not await base_boarding_client.association_is_viable(
        settings.TSYS_BOARDING_PARTNER, payload["association"], merchant_id=application.merchant_id, application_id=application.id
    ):
        return _finalize_section_save(
            db,
            application,
            "application_setup",
            payload,
            {
                "ok": False,
                "data": None,
                "tsys_errors": [{
                    "field": "association",
                    "message": (
                        "This pricing tier only offers TransFreedom plans, which "
                        "self-service onboarding can't complete. Please choose a "
                        "different pricing tier."
                    ),
                }],
                "http_status": 0,
            },
        )

    # Persist the local snapshot BEFORE calling GP (§7.2 resume guarantee).
    crud.update_application(
        db,
        application,
        tsys_association=payload["association"],
        tsys_lead_source=payload["leadSource"],
        tsys_sales_rep=payload["salesRep"],
    )
    db.commit()

    if not settings.TSYS_BOARDING_ENABLED:
        return _bootstrap_not_ready_result(db, application, "application_setup")

    envelope = await base_boarding_client.create_application(
        association=payload["association"],
        lead_source=payload["leadSource"],
        sales_rep=payload["salesRep"],
        merchant_id=application.merchant_id, application_id=application.id,
    )

    if not envelope["ok"]:
        # A real merchant-supplied-values rejection now (not a HubWallet
        # config bug the way PRD-HWONB-002's blanket 500 assumed) — return
        # it the normal AC-07 way so the wizard can render field-level
        # errors inline (setError), matching every other section.
        return _finalize_section_save(db, application, "application_setup", payload, envelope)

    data = envelope["data"] or {}
    tsys_app_id = data.get("appId")
    if tsys_app_id is None:
        # Genuinely can't happen per GP's documented contract on a 201 — this
        # is a real config/integration problem with the FIXED fields
        # (partner/merchantApplicationType/sBank), not a merchant input
        # error, so it still gets the hard 500 + engineering alert PRD-002
        # established for that failure mode.
        logger.error(
            "_save_application_setup_section: create_application returned "
            "ok=True but no appId in response for application_id=%s: %s",
            application.id,
            data,
        )
        crud.update_application(db, application, status="error")
        db.commit()
        raise APIException(
            status_code=500,
            message="We're having trouble starting your application. Please try again shortly.",
            error="onboarding_bootstrap_failed",
        )

    crud.update_application(db, application, tsys_app_id=tsys_app_id, status="in_progress")
    logger.info(
        "_save_application_setup_section: created GP application appId=%s for application_id=%s",
        tsys_app_id,
        application.id,
    )
    db.commit()

    return _finalize_section_save(db, application, "application_setup", payload, envelope)


def _bootstrap_not_ready_result(
    db: Session, application: MerchantOnboardingApplication, section: str
) -> Optional[Dict[str, Any]]:
    """
    Shared graceful-degradation guard (reviewer finding #3) — standardizes
    the "GP boarding isn't live yet" posture across every section to match
    business/processing/addresses/accounts/owners/card_types: when
    `not settings.TSYS_BOARDING_ENABLED or application.tsys_app_id is None`,
    the local snapshot (already persisted by the caller before this is
    invoked) is kept, `section_state[section]` is marked **pushed=True**
    (see the fix note below) with an informational `_system` tsys_error, and
    a normal 200-shaped response is returned — never a raised exception.
    Returns None when GP boarding IS ready (caller should proceed to the
    real GP push); returns the SectionSaveResponse-shaped dict otherwise.

    Used by products/training_activation, which previously relied on a
    stricter, exception-raising, `tsys_app_id`-only check further up in
    `save_section()` — that outer guard has been removed in favor of this
    shared, section-state-aware helper so every section behaves the same
    way under this condition.

    FIX (sequencing trap): this used to set `pushed: False` here, which was
    harmless for products/training_activation/fees in isolation (they have
    no N-2 prerequisite of their own today) but was the exact same footgun
    that made `_save_business_section` block the whole wizard — any section
    that ever gains a downstream prerequisite check keyed off
    `section_state[section]["pushed"]` would inherit the same trap. `pushed`
    is now True in disabled mode across every section for this reason; the
    informational `_system` tsys_error is kept so the wizard can still show
    a "not pushed to Global Payments yet" note without that note ever
    blocking sequencing.
    """
    if settings.TSYS_BOARDING_ENABLED and application.tsys_app_id is not None:
        return None

    logger.warning(
        "_bootstrap_not_ready_result(%r): TSYS_BOARDING_ENABLED=%s tsys_app_id=%s — "
        "snapshot saved locally, GP push skipped (application_id=%s)",
        section,
        settings.TSYS_BOARDING_ENABLED,
        application.tsys_app_id,
        application.id,
    )
    section_state = dict(application.section_state or {})
    section_state[section] = {
        "pushed": True,
        "tsys_errors": [
            {
                "field": "_system",
                "message": "Global Payments boarding is not enabled in this environment.",
            }
        ],
        "pushed_at": None,
    }
    update_fields: Dict[str, Any] = {"section_state": section_state}
    if application.current_step == section:
        next_step = _NEXT_STEP.get(section)
        if next_step:
            update_fields["current_step"] = next_step
    crud.update_application(db, application, **update_fields)
    db.commit()

    return {
        "section": section,
        "pushed": True,
        "tsys_errors": section_state[section]["tsys_errors"],
        "current_step": application.current_step,
        "data": None,
    }


def _finalize_section_save(
    db: Session,
    application: MerchantOnboardingApplication,
    section: str,
    payload: Dict[str, Any],
    envelope: Dict[str, Any],
) -> Dict[str, Any]:
    """
    Shared success/failure bookkeeping for a section push — updates
    `section_state[section]`, advances `current_step` on success (never on
    failure), commits, and returns the SectionSaveResponse-shaped dict the
    router hands straight back to the wizard.
    """
    section_state = dict(application.section_state or {})
    now = datetime.now(timezone.utc)

    if envelope["ok"]:
        data = envelope["data"] or {}
        extra: Dict[str, Any] = {}
        if section == "products":
            # PRD-HWONB-009 §6 — GP generates `terminalNumber` on first
            # create; store it for any future PATCH.
            terminal_number = data.get("terminalNumber")
            if terminal_number is not None:
                extra["terminalNumber"] = terminal_number

        section_state[section] = {
            "pushed": True,
            "tsys_errors": None,
            "pushed_at": now.isoformat(),
            **extra,
        }
        update_fields: Dict[str, Any] = {"section_state": section_state}
        if application.current_step == section:
            next_step = _NEXT_STEP.get(section)
            if next_step:
                update_fields["current_step"] = next_step
        crud.update_application(db, application, **update_fields)
        db.commit()

        logger.info(
            "save_section(%r): pushed OK for application_id=%s (tsys_app_id=%s)",
            section,
            application.id,
            application.tsys_app_id,
        )
        return {
            "section": section,
            "pushed": True,
            "tsys_errors": None,
            "current_step": application.current_step,
            "data": extra or None,
        }

    # GP rejected the section — record the errors, do NOT advance current_step.
    section_state[section] = {
        "pushed": False,
        "tsys_errors": envelope["tsys_errors"],
        "pushed_at": now.isoformat(),
    }
    crud.update_application(db, application, section_state=section_state)
    db.commit()

    logger.info(
        "save_section(%r): GP rejected for application_id=%s (tsys_app_id=%s) tsys_errors=%s",
        section,
        application.id,
        application.tsys_app_id,
        envelope["tsys_errors"],
    )
    return {
        "section": section,
        "pushed": False,
        "tsys_errors": envelope["tsys_errors"],
        "current_step": application.current_step,
        "data": None,
    }


# ---------------------------------------------------------------------------
# PRD-HWONB-011 — Pricing & Fees (AC-14 rate-card hard blocker)
# ---------------------------------------------------------------------------


async def _save_fees_section(
    db: Session,
    application: MerchantOnboardingApplication,
    payload: Dict[str, Any],
    named_misc_fee_overrides: Optional[Dict[str, float]] = None,
) -> Dict[str, Any]:
    """
    PRD-HWONB-011 §3 / PRD-HWONB-001 §7.2 — fees section save.

    `payload` is the merchant-facing GP-shaped fragment from
    `schemas.fees.FeesRequest.to_gp_payload()` (already validated by the
    router's request model). Persists the section snapshot to `fees_data`
    TWICE:
      1. Before the GP push, with the merchant-submitted fields only —
         the "a mid-request browser close loses nothing worse than
         re-saving this section" guarantee (§7.2).
      2. After a successful push, with the FULL merged body that was
         actually transmitted (merchant fields + `_resolve_rate_card()`'s
         injected rate-card values) — so a support engineer or later audit
         can see exactly what GP received without re-deriving anything
         (PRD-011 §3's explicit requirement). `push_fees()`'s envelope
         carries this via its `sent_payload` key.

    PRD-HWONB-015 — N-2 gate: requires `training_activation` pushed first
    (the merchant's selected association — captured upfront at Application
    Setup, Step 1 — is what `push_fees()` below resolves the rate card
    against; there is no sane default to fall back to here).
    """
    missing = _missing_prerequisites(application, "fees")
    if missing:
        raise BadRequestError(message=_sequencing_error_message(missing))

    # 1) Pre-GP-call safety snapshot (merchant-submitted fields only).
    crud.update_application(db, application, fees_data=payload)
    db.commit()

    # Reviewer finding #3 — standardize on the same graceful-degradation
    # posture as business/processing/addresses/accounts/owners/card_types
    # (check BOTH `TSYS_BOARDING_ENABLED` and `tsys_app_id is None`) instead
    # of this section's previous stricter tsys_app_id-only 409 raise.
    degraded = _bootstrap_not_ready_result(db, application, "fees")
    if degraded is not None:
        return {
            "ok": True,
            "tsys_errors": degraded["tsys_errors"],
            "section_state": application.section_state.get("fees"),
        }

    # 2) Push to GP — push_fees() internally resolves + injects the
    #    rate-card fields via _resolve_rate_card() before POSTing (see that
    #    method's docstring for the full field-by-field derivation and its
    #    prominent CERT-flakiness note). PRD-HWONB-014 — association is the
    #    merchant's own selected pricing tier (gated as a prerequisite of
    #    this very section, so it's always set by the time we get here),
    #    not the platform-wide settings default.
    envelope = await base_boarding_client.push_fees(
        application.tsys_app_id,
        payload,
        association=application.tsys_association,
        business_data=application.business_data,
        card_types_data=application.card_types_data,
        products_data=application.products_data,
        named_misc_fee_overrides=named_misc_fee_overrides,
        merchant_id=application.merchant_id, application_id=application.id,
    )

    # 3) Re-snapshot with the fully-merged body that was actually sent
    #    (falls back to the merchant-only payload if push_fees somehow
    #    didn't attach sent_payload, e.g. it failed before building a body).
    sent_payload = envelope.get("sent_payload") or payload
    crud.update_application(db, application, fees_data=sent_payload)

    section_state = dict(application.section_state or {})
    now_iso = datetime.now(timezone.utc).isoformat()

    if envelope["ok"]:
        section_state["fees"] = {"pushed": True, "tsys_errors": None, "pushed_at": now_iso}
        update_fields: Dict[str, Any] = {"section_state": section_state}
        if application.current_step == "fees":
            # Fees (Step 9) is followed by Documents (Step 10) per the
            # 11-step wizard order (PRD-HWONB-001 §0).
            update_fields["current_step"] = "documents"
        crud.update_application(db, application, **update_fields)
    else:
        section_state["fees"] = {
            "pushed": False,
            "tsys_errors": envelope["tsys_errors"],
            "pushed_at": None,
        }
        crud.update_application(db, application, section_state=section_state)

    db.commit()

    return {
        "ok": envelope["ok"],
        "tsys_errors": envelope["tsys_errors"],
        "section_state": section_state["fees"],
    }


def _sync_documents_section_state(db: Session, application: MerchantOnboardingApplication) -> None:
    """
    Documents has no `PUT /onboarding/{section}` shape (it's a separate
    upload/list/delete flow, PRD-HWONB-012), so section_state["documents"] is
    kept in sync here instead of via save_section(). `pushed=True` means SMA
    (the one unconditionally-required doc type, §2.2/AC-2) has been
    successfully uploaded — that is exactly the "Docs & Signing" condition
    PRD-HWONB-013 §1.2's transmit gate checks.
    """
    docs = crud.list_documents(db, application.id)
    # `pushed` must reflect GP acceptance, not merely a local row — the row is
    # created before the GP call and kept (with tsys_errors) on a GP rejection,
    # so an existence check would flip pushed=True for an SMA GP actually
    # refused (e.g. "Proof of 501(c)(3) Status required."), fooling the transmit
    # gate into a doomed GP call. tsys_synced_at is set only in upload_document's
    # ok branch, so it is the reliable "GP accepted this upload" signal.
    sma_uploaded = any(
        d.doc_type == "SMA" and d.deleted_at is None and d.tsys_synced_at is not None
        for d in docs
    )

    section_state = dict(application.section_state or {})
    section_state["documents"] = {
        "pushed": sma_uploaded,
        "tsys_errors": None,
        "pushed_at": datetime.now(timezone.utc).isoformat() if sma_uploaded else None,
    }
    crud.update_application(db, application, section_state=section_state)


def _apply_upload_envelope(
    db: Session, document: MerchantOnboardingDocument, envelope: Dict[str, Any]
) -> None:
    """Record a GP upload_attachment result on a document row. Single home for
    the attachmentId-nesting extraction (GP nests it at
    attachments[0].attachmentId, C-22) so both the initial upload and the
    resubmit sweep stay consistent."""
    if envelope["ok"]:
        data = envelope["data"] or {}
        attachments = data.get("attachments") or []
        attachment_id = attachments[0].get(field_mapping.ATTACHMENT_ID_FIELD) if attachments else None
        crud.update_document(
            db,
            document,
            tsys_upload_ref=str(attachment_id) if attachment_id is not None else None,
            tsys_synced_at=datetime.now(timezone.utc),
            tsys_errors=None,
        )
    else:
        crud.update_document(db, document, tsys_errors=envelope["tsys_errors"])


def _is_charity_501c3(application: MerchantOnboardingApplication) -> bool:
    """True when the saved Business section flagged the merchant as a 501(c)(3)
    charity — the sole trigger for GP's CH-bundling requirement below."""
    return bool(
        ((application.business_data or {}).get("businessDetails") or {}).get("charity501c3Exempt") is True
    )


def _read_document_bytes(db: Session, application: MerchantOnboardingApplication, doc) -> Optional[bytes]:
    """Re-read a stored document's bytes (mirrors upload_document's storage-path
    reconstruction) so a doc can be (re)submitted to GP after the fact."""
    import os

    from src.apps.base.services import get_active_cdn
    from src.apps.files.helper.io import download_file_from_s3, download_file_from_system_path

    active_cdn = get_active_cdn(db)
    base_path = active_cdn.upload_path if active_cdn else f"{os.getcwd()}/{settings.UPLOADS_DIR}"
    full_path = base_path + "/" + doc.s3_key
    if active_cdn and active_cdn.label.lower() == "s3":
        return download_file_from_s3(path=full_path)
    return download_file_from_system_path(path=full_path)


async def _submit_pending_documents(db: Session, application: MerchantOnboardingApplication) -> None:
    """(Re)submit any document row GP hasn't accepted yet (tsys_synced_at IS NULL).

    Charity501c3 apps: GP rejects a standalone SMA/ACH upload with 40251
    "Proof of 501(c)(3) Status required." — even when a CH attachment already
    exists — and only accepts them when the CH (proof) file rides along in the
    SAME multipart call (`?docType=SMA&docType=CH`, positional). So for a
    charity app we bundle every pending non-CH doc together with the CH file in
    one combined call. Non-charity apps just re-submit each pending doc
    standalone (self-heals transient GP failures). Runs after every upload, so
    the CH upload is what clears previously-pending SMA/ACH — upload order is
    irrelevant.

    Calls the client directly (not upload_document) — no recursion, no
    re-running of storage/validation.
    """
    docs = [d for d in crud.list_documents(db, application.id) if d.tsys_synced_at is None]
    if not docs:
        return

    if not _is_charity_501c3(application):
        for doc in docs:
            file_bytes = _read_document_bytes(db, application, doc)
            if not file_bytes:
                logger.warning("submit_pending: no stored bytes for document %s (%s)", doc.id, doc.s3_key)
                continue
            envelope = await base_boarding_client.upload_attachment(
                application.tsys_app_id, doc.doc_type, file_bytes, doc.file_name, doc.content_type,
                merchant_id=application.merchant_id, application_id=application.id,
            )
            _apply_upload_envelope(db, doc, envelope)
        return

    # Charity branch — need a CH proof file to bundle with the pending non-CH docs.
    ch_doc = next(
        (d for d in crud.list_documents(db, application.id) if d.doc_type == "CH" and d.deleted_at is None),
        None,
    )
    pending = [d for d in docs if d.doc_type != "CH"]
    if ch_doc is None or not pending:
        return  # CH not uploaded yet (or nothing pending) — its upload will re-trigger this sweep

    ch_bytes = _read_document_bytes(db, application, ch_doc)
    if not ch_bytes:
        logger.warning("submit_pending: no stored bytes for CH document %s — cannot bundle", ch_doc.id)
        return

    parts, submitted = [], []
    for doc in pending:
        file_bytes = _read_document_bytes(db, application, doc)
        if not file_bytes:
            logger.warning("submit_pending: no stored bytes for document %s (%s)", doc.id, doc.s3_key)
            continue
        parts.append((doc.doc_type, doc.file_name, doc.content_type, file_bytes))
        submitted.append(doc)
    if not parts:
        return
    # CH rides along last so GP's proof requirement is satisfied for the others.
    parts.append(("CH", ch_doc.file_name, ch_doc.content_type, ch_bytes))

    envelope = await base_boarding_client.upload_attachments(
        application.tsys_app_id, parts, merchant_id=application.merchant_id, application_id=application.id,
    )
    if not envelope["ok"]:
        for doc in submitted:
            crud.update_document(db, doc, tsys_errors=envelope["tsys_errors"])
        return
    # GP returns attachments positionally; the trailing one is the (duplicate) CH — ignore it,
    # the local CH row is already synced from its own upload.
    returned = (envelope["data"] or {}).get("attachments") or []
    for i, doc in enumerate(submitted):
        att = returned[i] if i < len(returned) else None
        att_id = att.get(field_mapping.ATTACHMENT_ID_FIELD) if att else None
        crud.update_document(
            db, doc,
            tsys_upload_ref=str(att_id) if att_id is not None else None,
            tsys_synced_at=datetime.now(timezone.utc),
            tsys_errors=None,
        )
    # ponytail: creates one duplicate CH attachment in GP per sweep (GP tolerates it; transmit
    # only needs presence). Collapse to a single batched submit if dup noise ever matters.


async def upload_document(
    db: Session,
    application: MerchantOnboardingApplication,
    doc_type: str,
    file_bytes: bytes,
    file_name: str,
    content_type: str,
) -> MerchantOnboardingDocument:
    """
    POST /onboarding/documents — PRD-HWONB-012 §2.

    Validates doc_type against the 15-code enum (C-20 — `Tax`, not `TAX`),
    the content-type/size allowlist (§2.3 — 2MB production limit, not
    CERT's looser 10MB), and the business-information ordering gate (§2.4 —
    GP itself rejects attachment uploads until business is saved). Persists
    the local row + uploads to storage BEFORE calling GP (PRD-HWONB-001
    §7.2 — a browser close mid-request must never lose more than "this
    upload needs retrying"), then calls GP and records the result
    (tsys_upload_ref / tsys_errors) on the same row rather than raising on a
    GP-side rejection — a partial/GP-rejected upload is visible to the
    wizard via tsys_errors, not silently dropped.
    """
    if doc_type not in field_mapping.DOC_TYPE_CHOICES:
        raise APIException(
            status_code=422,
            message=f"Unknown document type {doc_type!r}.",
            error="onboarding_invalid_doc_type",
        )

    if not _section_pushed(application.section_state, field_mapping.DOCUMENTS_PREREQUISITE_SECTION):
        # Mirrors GP's own exact wording (§2.4) so the wizard's message reads
        # the same whether the block came from this local pre-check or a
        # real GP 40008.
        raise APIException(
            status_code=409,
            message="Please complete business information.",
            error="onboarding_business_not_pushed",
        )

    if not file_bytes:
        raise APIException(status_code=422, message="The file is empty.", error="onboarding_empty_file")

    if content_type not in field_mapping.ALLOWED_DOCUMENT_CONTENT_TYPES:
        raise APIException(
            status_code=422,
            message="Unsupported file type. Allowed formats: PDF, JPG, XLS, XLSX.",
            error="onboarding_invalid_file_type",
        )

    # SECURITY (M-2): `content_type` above is the client-supplied multipart
    # header — fully attacker-controlled and not proof of the actual file
    # contents. Independently verify the first bytes (and, for XLSX, the zip
    # member list) match the declared type before trusting it any further.
    if not content_matches_declared_type(file_bytes, content_type):
        raise APIException(
            status_code=422,
            message="The file contents do not match the declared file type.",
            error="onboarding_file_content_mismatch",
        )

    if len(file_bytes) > field_mapping.MAX_DOCUMENT_SIZE_BYTES:
        raise APIException(
            status_code=422,
            message="File exceeds the maximum size of 2MB.",
            error="onboarding_file_too_large",
        )

    if application.tsys_app_id is None:
        raise APIException(
            status_code=409,
            message="This application has not been started with Global Payments yet.",
            error="onboarding_not_bootstrapped",
        )

    # Store the file first (mirrors src/apps/files' upload_file_single/
    # upload_file_to_s3 pattern) so a subsequent GP-call failure never loses
    # the file itself — only the GP sync (tsys_upload_ref) needs retrying.
    import os

    from src.apps.base.services import get_active_cdn
    from src.apps.files.helper.io import (
        _generate_unique_filename,
        upload_file_to_s3,
        upload_file_to_system_path,
    )

    active_cdn = get_active_cdn(db)
    unique_filename, _basename, _fmt = _generate_unique_filename(file_name)
    serve_path = f"/onboarding_documents/{application.id}/{unique_filename}"
    base_path = active_cdn.upload_path if active_cdn else f"{os.getcwd()}/{settings.UPLOADS_DIR}"
    upload_path = base_path + serve_path

    if active_cdn and active_cdn.label.lower() == "s3":
        uploaded = upload_file_to_s3(file_content=file_bytes, content_type=content_type, path=upload_path)
    else:
        uploaded = upload_file_to_system_path(file_content=file_bytes, path=upload_path)

    if not uploaded:
        raise APIException(
            status_code=500,
            message="Could not store the uploaded document. Please try again.",
            error="onboarding_document_storage_failed",
        )

    s3_key = serve_path.lstrip("/")
    document = crud.create_document_row(
        db,
        application.id,
        doc_type=doc_type,
        file_name=file_name,
        content_type=content_type,
        size_bytes=len(file_bytes),
        s3_key=s3_key,
    )
    db.commit()

    # Charity501c3 apps: a standalone SMA/ACH upload is always rejected by GP
    # ("Proof of 501(c)(3) Status required.") — it only succeeds bundled with
    # the CH file in one call. So don't call GP standalone here for a non-CH
    # doc; leave the row pending and let _submit_pending_documents bundle it
    # with the CH (which itself uploads fine standalone). Non-charity apps and
    # the CH doc take the normal standalone path.
    if _is_charity_501c3(application) and doc_type != "CH":
        logger.info(
            "upload_document: deferring charity non-CH doc %s (doc_type=%s) for bundled submit",
            document.id, doc_type,
        )
    else:
        envelope = await base_boarding_client.upload_attachment(
            application.tsys_app_id, doc_type, file_bytes, file_name, content_type, merchant_id=application.merchant_id, application_id=application.id
        )
        if not envelope["ok"]:
            logger.warning(
                "upload_document: GP upload_attachment failed for application %s doc_type=%s errors=%s",
                application.id,
                doc_type,
                envelope["tsys_errors"],
            )
        _apply_upload_envelope(db, document, envelope)

    # A successful upload (esp. the CH proof) can unblock docs GP hasn't
    # accepted yet, so (re)submit every still-pending doc — bundling with CH
    # for charity apps.
    await _submit_pending_documents(db, application)

    _sync_documents_section_state(db, application)
    db.commit()
    return document


async def list_documents(
    db: Session, application: MerchantOnboardingApplication
) -> Dict[str, Any]:
    """
    GET /onboarding/documents — PRD-HWONB-012 §4 ("for resume/audit,
    reconciling local rows against GP's list").

    Local rows are the source of truth for the response (the wizard needs a
    fast, reliable list even if GP is briefly unreachable); the GP list call
    is best-effort reconciliation only — logged on mismatch, never raised,
    since this endpoint must not fail just because an audit comparison
    couldn't run.
    """
    documents = crud.list_documents(db, application.id)

    if application.tsys_app_id is not None:
        try:
            envelope = await base_boarding_client.list_attachments(application.tsys_app_id, merchant_id=application.merchant_id, application_id=application.id)
            if envelope["ok"]:
                gp_data = envelope["data"] or {}
                gp_attachments = gp_data.get("attachments") or []
                gp_ids = {str(a.get("attachmentId")) for a in gp_attachments if a.get("attachmentId") is not None}
                local_refs = {d.tsys_upload_ref for d in documents if d.tsys_upload_ref}
                if gp_ids != local_refs:
                    logger.warning(
                        "list_documents: local/GP attachment mismatch for application %s — "
                        "local_refs=%s gp_ids=%s (audit only, not corrected automatically)",
                        application.id,
                        local_refs,
                        gp_ids,
                    )
            else:
                logger.info(
                    "list_documents: GP list_attachments failed for application %s errors=%s",
                    application.id,
                    envelope["tsys_errors"],
                )
        except Exception as exc:  # best-effort audit only — never blocks the response
            logger.warning("list_documents: GP reconciliation call raised for application %s: %s", application.id, exc)

    sma_uploaded = _section_pushed(application.section_state, "documents")
    return {"documents": documents, "sma_uploaded": sma_uploaded}


async def delete_document(
    db: Session, application: MerchantOnboardingApplication, document_id: int
) -> MerchantOnboardingDocument:
    """
    DELETE /onboarding/documents/{id} — PRD-HWONB-012 §2.5, pre-transmit
    only (§4/§5 of the PRD-013 review screen implies documents are locked
    once transmitted, matching every other section's edit-lock after
    transmit).
    """
    if application.status not in ("draft", "in_progress", "ready_to_transmit"):
        raise APIException(
            status_code=409,
            message="Documents can no longer be modified after the application has been transmitted.",
            error="onboarding_documents_locked",
        )

    document = crud.get_document(db, application.id, document_id)
    if document is None:
        raise APIException(status_code=404, message="Document not found.", error="onboarding_document_not_found")

    if document.tsys_upload_ref and application.tsys_app_id is not None:
        try:
            gp_attachment_id = int(document.tsys_upload_ref)
        except (TypeError, ValueError):
            gp_attachment_id = None
            logger.warning(
                "delete_document: non-numeric tsys_upload_ref %r on document %s — skipping GP delete call",
                document.tsys_upload_ref,
                document_id,
            )
        if gp_attachment_id is not None:
            envelope = await base_boarding_client.delete_attachment(
                application.tsys_app_id, gp_attachment_id, merchant_id=application.merchant_id, application_id=application.id
            )
            if not envelope["ok"]:
                logger.warning(
                    "delete_document: GP delete_attachment failed for application %s document %s errors=%s — "
                    "deleting the local row anyway so the wizard doesn't get stuck on a GP-side hiccup",
                    application.id,
                    document_id,
                    envelope["tsys_errors"],
                )

    crud.soft_delete_document(db, document)
    _sync_documents_section_state(db, application)
    db.commit()
    return document


async def get_uma(application: MerchantOnboardingApplication) -> bytes:
    """
    GET /onboarding/uma — PRD-HWONB-012 §1. Proxies GP's binary agreement
    PDF; idempotent/re-callable (not a "generate once" action — GP returns
    the current state of the agreement every time).
    """
    if application.tsys_app_id is None:
        raise APIException(
            status_code=409,
            message="This application has not been started with Global Payments yet.",
            error="onboarding_not_bootstrapped",
        )

    envelope = await base_boarding_client.get_uma(application.tsys_app_id, merchant_id=application.merchant_id, application_id=application.id)
    if not envelope["ok"]:
        raise APIException(
            status_code=502,
            message="Could not generate the merchant agreement right now. Please try again shortly.",
            error={"tsys_errors": envelope["tsys_errors"]},
        )

    data = envelope["data"]
    if not isinstance(data, (bytes, bytearray)):
        logger.error("get_uma: unexpected non-binary UMA response for application %s: %r", application.id, type(data))
        raise APIException(
            status_code=502,
            message="Global Payments returned an unexpected response for the merchant agreement.",
            error="onboarding_uma_bad_response",
        )
    return bytes(data)


async def transmit(db: Session, application: MerchantOnboardingApplication) -> Dict[str, Any]:
    """
    POST /onboarding/transmit — PRD-HWONB-013 §1.

    Re-validates all 9 GP-required sections (§1.2's exact list) PLUS
    `business` (belt-and-suspenders — GP's own error text never names
    Business since every other section already can't be pushed without it,
    N-2) from local `section_state`/dedicated-table counts — never by
    re-querying GP (§1.3) — so an incomplete application never wastes a
    real transmit attempt. A duplicate transmit on an application already
    past the pre-transmit statuses is treated as an idempotent no-op
    success (§1.2/AC-2), and — because the event is only dispatched on the
    one call that actually performs the status transition — the
    `onboarding_submitted` email guarded by `submitted_email_sent_at`
    (T-09) can never fire twice even under a retried request.
    """
    if application.status not in ("draft", "in_progress", "ready_to_transmit"):
        logger.info(
            "transmit: application %s already at status=%r — idempotent no-op (AC-2)",
            application.id,
            application.status,
        )
        return {"transmitted": True, "already_transmitted": True, "status": application.status}

    section_state = application.section_state or {}

    if not _section_pushed(section_state, "business"):
        raise APIException(
            status_code=422,
            message="Please complete business information.",
            error="onboarding_business_not_pushed",
        )

    missing: List[str] = []
    for key, label in _TRANSMIT_REQUIRED_SECTIONS:
        if key == "accounts":
            ok = crud.count_accounts(db, application.id) >= 1
        elif key == "owners":
            ok = crud.count_owners(db, application.id) >= 1
        else:
            ok = _section_pushed(section_state, key)
        if not ok:
            missing.append(label)

    if missing:
        # Mirrors GP's own exact wording (§1.2) so the wizard's error banner
        # reads the same whether the block came from this local pre-check
        # or a real GP 40008.
        raise APIException(
            status_code=422,
            message=(
                f"Application Failed Transmission. {', '.join(missing)} "
                "are required to transmit the application"
            ),
            error="onboarding_transmit_incomplete",
        )

    if application.tsys_app_id is None:
        raise APIException(
            status_code=409,
            message="This application has not been started with Global Payments yet.",
            error="onboarding_not_bootstrapped",
        )

    # Local pre-check passed — walk the state machine the rest of the way to
    # "transmitted" before calling GP. `in_progress -> ready_to_transmit` is
    # exactly the transition the state machine models for "all sections are
    # now pushed" (helpers/status_machine.py), done here rather than inside
    # save_section() since documents (the last section) has no save_section
    # hook of its own (§ per this module's docstring).
    if application.status == "in_progress":
        status_machine.assert_transition(application.status, "ready_to_transmit")
        crud.update_application(db, application, status="ready_to_transmit")
        provider_services.sync_tsys_onboarding_status(application.merchant_id, application.status, db)
        db.flush()
    status_machine.assert_transition(application.status, "transmitted")

    envelope = await base_boarding_client.transmit(application.tsys_app_id, merchant_id=application.merchant_id, application_id=application.id)

    if not envelope["ok"]:
        errors = envelope["tsys_errors"] or []
        # §1.2 — a re-transmit of an already-transmitted appId is not a
        # GP-error-shaped failure in spirit; `errorCode` itself is never
        # assumed numeric (AC-4 — a real example returns the opaque string
        # "SYSTEM_ERROR_DOWNSTREAM").
        already_transmitted = any(
            "already transmitted" in str(e.get("message", "")).lower() for e in errors
        )
        if not already_transmitted:
            logger.error(
                "transmit: GP transmit failed for application %s tsys_app_id=%s errors=%s",
                application.id,
                application.tsys_app_id,
                errors,
            )
            db.commit()  # persist the ready_to_transmit bump even though GP itself rejected
            raise APIException(
                status_code=502,
                message="Global Payments rejected the transmit request.",
                error={"tsys_errors": errors},
            )
        logger.info(
            "transmit: GP reports appId=%s already transmitted — treating as idempotent success (AC-2)",
            application.tsys_app_id,
        )

    tsys_mid = (envelope.get("data") or {}).get("merchantId")
    update_fields = {"status": "transmitted", "current_step": "review"}
    if tsys_mid and not application.tsys_mid:
        update_fields["tsys_mid"] = str(tsys_mid)
    crud.update_application(db, application, **update_fields)
    provider_services.sync_tsys_onboarding_status(
        application.merchant_id, application.status, db, tsys_mid=application.tsys_mid
    )
    db.commit()

    try:
        await EventDispatcher.dispatch(
            BaseEvent(
                event_type="onboarding.transmitted",
                data={
                    "application_id": application.id,
                    "merchant_id": application.merchant_id,
                    "tsys_app_id": application.tsys_app_id,
                },
            )
        )
    except Exception as exc:
        logger.warning(
            "transmit: failed to dispatch onboarding.transmitted for application %s: %s",
            application.id,
            exc,
        )

    return {"transmitted": True, "already_transmitted": False, "status": application.status}


async def refresh_status(db: Session, application: MerchantOnboardingApplication) -> Dict[str, Any]:
    """
    GET /onboarding/status (manual refresh) + the nightly poll task's core
    logic — PRD-HWONB-013 §2.

    Calls both GP endpoints per §2.3: `get_status()` for the coarse
    `applicationStatus`/`mid` (state-machine input) and, once a MID exists,
    `get_activity()` for the enumerated progress-UI data. The FULL raw
    payload from both calls is persisted to `tsys_decision_payload`
    regardless of what `_map_gp_status_to_local()` concludes (its own
    module docstring / C-24) so a real CERT-observed decision can be
    reconciled against this mapping later. `onboarding.approved` /
    `onboarding.rejected` are only dispatched when this call is the one
    that actually flips the local status — every later poll of an
    already-`approved`/`rejected` application maps to the same status again
    and dispatches nothing, so this is idempotent by construction without
    needing a separate "already dispatched" flag (the per-application email
    guards in listener.py are a second, independent idempotency layer).
    """
    if application.tsys_app_id is None:
        raise APIException(
            status_code=409,
            message="This application has not been started with Global Payments yet.",
            error="onboarding_not_bootstrapped",
        )

    status_envelope = await base_boarding_client.get_status(application.tsys_app_id, merchant_id=application.merchant_id, application_id=application.id)
    activity_envelope: Optional[Dict[str, Any]] = None
    if application.tsys_mid:
        activity_envelope = await base_boarding_client.get_activity(application.tsys_mid, merchant_id=application.merchant_id, application_id=application.id)

    raw_payload: Dict[str, Any] = {
        "status": status_envelope["data"] if status_envelope["ok"] else {"tsys_errors": status_envelope["tsys_errors"]},
        "activity": (
            activity_envelope["data"]
            if activity_envelope is not None and activity_envelope["ok"]
            else None
        ),
        "polled_at": datetime.now(timezone.utc).isoformat(),
    }

    tsys_decision_payload = dict(application.tsys_decision_payload or {})
    tsys_decision_payload["last_poll"] = raw_payload
    now = datetime.now(timezone.utc)

    gp_status_row: Dict[str, Any] = {}
    if status_envelope["ok"] and isinstance(status_envelope["data"], dict):
        dashboard_list = status_envelope["data"].get("dashboardDTOList") or []
        # Pre-transmit (and per C-24, observed CERT-side) this list is empty —
        # nothing to map yet, leave the local status untouched this poll.
        if dashboard_list:
            gp_status_row = dashboard_list[0] or {}

    previous_status = application.status
    new_local_status = status_machine._map_gp_status_to_local(
        gp_status_row, current_local_status=previous_status
    )

    update_fields: Dict[str, Any] = {
        "last_status_polled_at": now,
        "tsys_decision_payload": tsys_decision_payload,
    }

    raw_status_string = (
        gp_status_row.get("applicationStatus")
        or gp_status_row.get("approvalStatus")
        or gp_status_row.get("merchantBoardingStatus")
    )
    if raw_status_string:
        update_fields["last_tsys_status"] = str(raw_status_string)

    mid_from_status = gp_status_row.get("mid")
    if mid_from_status and not application.tsys_mid:
        update_fields["tsys_mid"] = str(mid_from_status)

    transitioned_to: Optional[str] = None
    if new_local_status != previous_status and status_machine.can_transition(previous_status, new_local_status):
        update_fields["status"] = new_local_status
        transitioned_to = new_local_status

    crud.update_application(db, application, **update_fields)
    if "status" in update_fields:
        provider_services.sync_tsys_onboarding_status(application.merchant_id, application.status, db)
    db.commit()

    if transitioned_to == "approved":
        try:
            await EventDispatcher.dispatch(
                BaseEvent(
                    event_type="onboarding.approved",
                    data={
                        "application_id": application.id,
                        "merchant_id": application.merchant_id,
                        "tsys_app_id": application.tsys_app_id,
                        "tsys_mid": application.tsys_mid,
                    },
                )
            )
        except Exception as exc:
            logger.warning(
                "refresh_status: failed to dispatch onboarding.approved for application %s: %s",
                application.id,
                exc,
            )
    elif transitioned_to == "rejected":
        try:
            await EventDispatcher.dispatch(
                BaseEvent(
                    event_type="onboarding.rejected",
                    data={
                        "application_id": application.id,
                        "merchant_id": application.merchant_id,
                        "tsys_app_id": application.tsys_app_id,
                        "reason": raw_status_string,
                    },
                )
            )
        except Exception as exc:
            logger.warning(
                "refresh_status: failed to dispatch onboarding.rejected for application %s: %s",
                application.id,
                exc,
            )

    return {
        "status": application.status,
        "last_tsys_status": application.last_tsys_status,
        "last_status_polled_at": application.last_status_polled_at,
        "activity": raw_payload["activity"],
        "transitioned_to": transitioned_to,
    }


async def restart_after_rejection(db: Session, merchant) -> MerchantOnboardingApplication:
    """
    POST /onboarding/restart — PRD-HWONB-013 §3. Only valid when the
    merchant's current application `status == "rejected"`.

    Soft-deletes the rejected application row, then calls
    `start_or_resume()` again to create a brand-new row with a clean
    `tsys_app_id=None` and `current_step="application_setup"` — the
    merchant is routed back through Application Setup, which mints a
    genuinely new `appId` via `create_application()` once they resubmit it
    (PRD-HWONB-015). PRD-HWONB-002 AC-5 / PRD-HWONB-013 AC-5 are explicit
    this must NEVER reuse the rejected application's `appId`.
    Resubmission-after-pend (`POST .../ucm/pend/resubmit`) is out of scope
    for v1 (PRD-HWONB-013 §3) — this is the only path back from `rejected`.
    """
    application = crud.get_active_application(db, merchant.id)
    if application is None or application.status != "rejected":
        raise APIException(
            status_code=409,
            message="Restart is only available for a rejected application.",
            error="onboarding_restart_not_allowed",
        )

    crud.soft_delete_application(db, application)
    db.commit()

    new_application = await start_or_resume(db, merchant)
    # Without this, the merchant's provider config stays stuck on "rejected"
    # (mirrored there when the old application was rejected) until the new
    # application transitions again — reset it to reflect the fresh restart.
    provider_services.sync_tsys_onboarding_status(merchant.id, new_application.status, db)
    db.commit()
    return new_application


async def provision_transit(db: Session, application: MerchantOnboardingApplication) -> None:
    """
    Called from listener.py's `onboarding.approved` handler once GP issues a
    MID. Provisions the merchant's TSYS TransIT processing credentials so
    they can start taking payments immediately.

    Reuses (does not reimplement) `admin_upsert_tsys_credentials()` from
    `src/apps/payment_providers/services.py` — that function already knows
    how to call `TSYSProvider._generate_transaction_key()` and persist
    encrypted credentials. That function's request schema
    (`AdminUpsertTsysCredentialsRequest`) normally carries admin-entered,
    per-merchant `user_id`/`password`/`developer_id`/`device_id` — this
    onboarding flow has no such per-merchant values (GP's Base Boarding
    approval only ever hands back a MID, never TransIT gateway
    credentials), so this reuses the platform-level GenerateKey fallback
    settings that already exist for exactly this "merchant has no
    active_provider_id yet" case: `TSYS_USER_ID` / `TSYS_PASSWORD` /
    `TSYS_DEVELOPER_ID` / `TSYS_DEVICE_ID` (src/core/config.py — explicitly
    documented as "Default ... (fallback)", and distinct from the
    `HW_TSYS_*` settings, which are HubWallet's own platform settlement
    account, not for provisioning merchants). If those aren't configured on
    this deployment, this raises rather than fabricating credentials —
    an admin can still finish provisioning manually via the existing
    `PUT /admin/merchants/{id}/tsys/credentials` endpoint; listener.py's
    outer exception handler logs this without crashing the poll/refresh
    call that triggered it.
    """
    if not application.tsys_mid:
        raise APIException(
            status_code=409,
            message="Cannot provision processing credentials before Global Payments has issued a MID.",
            error="onboarding_no_mid",
        )

    if not (
        settings.TSYS_USER_ID
        and settings.TSYS_PASSWORD
        and settings.TSYS_DEVELOPER_ID
        and settings.TSYS_DEVICE_ID
    ):
        logger.error(
            "provision_transit: platform TSYS_USER_ID/TSYS_PASSWORD/TSYS_DEVELOPER_ID/"
            "TSYS_DEVICE_ID fallback credentials are not configured — cannot auto-provision "
            "application %s (tsys_mid=%s). An admin must upsert TSYS credentials manually "
            "via PUT /admin/merchants/{id}/tsys/credentials.",
            application.id,
            application.tsys_mid,
        )
        raise APIException(
            status_code=503,
            message="TSYS auto-provisioning is not configured on this platform.",
            error="onboarding_tsys_not_configured",
        )

    from src.apps.payment_providers.schemas.requests import AdminUpsertTsysCredentialsRequest
    from src.apps.payment_providers.services import admin_upsert_tsys_credentials

    payload = AdminUpsertTsysCredentialsRequest(
        merchant_id=application.tsys_mid,
        device_id=settings.TSYS_DEVICE_ID,
        developer_id=settings.TSYS_DEVELOPER_ID,
        user_id=settings.TSYS_USER_ID,
        password=settings.TSYS_PASSWORD,
    )

    await admin_upsert_tsys_credentials(application.merchant_id, payload, db)
    logger.info(
        "provision_transit: TSYS TransIT credentials provisioned for application %s (tsys_mid=%s)",
        application.id,
        application.tsys_mid,
    )