feat(transport): report_dispense RPC — capture a cash-out on the machine's report, not on payment (ADR-005 §1–§2)

The structural fix for bitspire#122. _handle_payment used to spawn
process_settlement the instant a cash_out payment landed — before the
machine had begun to dispense — so a jam two seconds later found the
legs already paid and `processed` was the honest answer. Payment is now
the authorization; the machine's report is the capture.

A cash_out lands as awaiting_dispense and is not distributed. The new
`report_dispense` handler (identity from the VERIFIED transport sender,
same as create_withdraw / get_machine_config) stores every report
append-only and moves the settlement: dispense_confirmed → pending and
distribution runs; some notes out → partial_pending, held whole
(ADR-005 Decision 1, one distribution when the shortfall is resolved);
nothing out → cash_owed, first on the worklist. A report naming
remediates_txid moves the owed settlement it names to pending in full.
Already-captured settlements are recorded but never moved — a report
cannot un-pay legs. A byte-identical resend is acked without a new row.

Both orders of arrival are handled: a report that precedes its payment
(hold invoices settle after the dispense; the invoice listener can lag)
is stored unlinked and adopted when the settlement is inserted, through
the same transition. counts_uncertain on a report mirrors onto the
machine immediately rather than at the next heartbeat. The state-event
consumer mirrors cash_out_held_* onto dca_machines, including clearing it.

Soft-fails like the other RPCs: without register_rpc the settlements sit
in awaiting_dispense and surface as dispense_unreported — the honest state.

tasks.py

@@ -169,8 +169,17 @@ async def _handle_payment(payment: Payment) -> None:
     if isinstance(nostr_event_id, str) and nostr_event_id:
         data.bitspire_event_id = nostr_event_id
 
-    # 3) Insert + distribute.
-    settlement = await create_settlement_idempotent(data, initial_status="pending")
+    # 3) Insert. ADR-005 §1: payment is authorization, the machine's dispense
+    #    report is capture. A cash_out waits in `awaiting_dispense` for that
+    #    report (handled in dispense_transport); distribution runs only once it
+    #    says dispense_confirmed. A cash_in has no dispense and proceeds as
+    #    before. Before this gate the legs were paid sub-second, before the
+    #    machine had even begun to dispense — which is how a jam read
+    #    `processed` on 2026-10-09 (bitspire#122).
+    is_cash_out = data.tx_type == "cash_out"
+    settlement = await create_settlement_idempotent(
+        data, initial_status="awaiting_dispense" if is_cash_out else "pending"
+    )
     if settlement is None:
         logger.error(
             f"spirekeeper: failed to insert settlement for "
@@ -185,6 +194,10 @@ async def _handle_payment(payment: Payment) -> None:
         f"(super_fee={data.platform_fee_sats} "
         f"operator_fee={data.operator_fee_sats})"
     )
+    if is_cash_out:
+        await _await_dispense_or_adopt(settlement, machine, data)
+        return
+
     # Spawn distribution on a background task so the LNbits invoice queue
     # (shared across all extensions) keeps draining while we move sats.
     # Concurrency-safe: process_settlement uses claim_settlement_for_processing
@@ -196,6 +209,23 @@ async def _handle_payment(payment: Payment) -> None:
     task.add_done_callback(_inflight_distributions.discard)
 
 
+async def _await_dispense_or_adopt(
+    settlement, machine: Machine, data: CreateDcaSettlementData
+) -> None:
+    """A cash_out waits for the machine's dispense report (ADR-005 §1) — unless
+    the report is already here. Under hold invoices the payment settles AFTER
+    the dispense, and the invoice listener can lag the transport, so an orphan
+    report for this txid is adopted and applied now."""
+    if settlement.status != "awaiting_dispense" or not data.bitspire_txid:
+        return
+    from .crud import get_latest_unlinked_dispense_report
+    from .dispense_transport import adopt_unlinked_report
+
+    early = await get_latest_unlinked_dispense_report(machine.id, data.bitspire_txid)
+    if early is not None:
+        await adopt_unlinked_report(settlement, machine, early)
+
+
 async def _record_rejected(payment: Payment, machine: Machine, exc: Exception) -> None:
     """Insert a minimal `dca_settlements` row with `status='rejected'` and
     the exception message for operator forensics.
@@ -438,12 +468,38 @@ async def _record_counts_uncertainty(
     await set_machine_counts_uncertain(machine_id, since)
 
 
+async def _record_cash_out_hold(
+    machine_id: str, payload, set_machine_cash_out_hold
+) -> None:
+    """Mirror the machine's cash-out hold (ADR-005 §5) onto its registry row.
+
+    Written on every state event, including when absent, because the machine
+    clearing the hold — after an operator recount or resume_cash_out — matters
+    exactly as much as it setting one.
+    """
+    from datetime import datetime as _datetime
+    from datetime import timezone as _timezone
+
+    since = None
+    if payload.cash_out_held_since is not None:
+        since = _datetime.fromtimestamp(
+            int(payload.cash_out_held_since), tz=_timezone.utc
+        )
+    await set_machine_cash_out_hold(
+        machine_id,
+        since,
+        payload.cash_out_held_reason if since else None,
+        payload.cash_out_held_code if since else None,
+    )
+
+
 async def _handle_cassette_state_event(
     event_message,
     get_machine_by_atm_pubkey_hex,
     apply_reported_state,
     mark_cassette_ops_acked,
     set_machine_counts_uncertain,
+    set_machine_cash_out_hold=None,
 ) -> None:
     """Verify signature, resolve the operator's signer, decrypt via the
     signer abstraction (bunker round-trip for RemoteBunkerSigner; direct
@@ -558,3 +614,8 @@ async def _handle_cassette_state_event(
     # _record_op_acknowledgements for why. Same for the uncertainty marker.
     await _record_op_acknowledgements(machine.id, payload, mark_cassette_ops_acked)
     await _record_counts_uncertainty(machine.id, payload, set_machine_counts_uncertain)
+    if set_machine_cash_out_hold is None:
+        from .crud import set_machine_cash_out_hold as _default_set_hold
+
+        set_machine_cash_out_hold = _default_set_hold
+    await _record_cash_out_hold(machine.id, payload, set_machine_cash_out_hold)