feat(v2): wire cash-in routing — direction discriminator + DCA skip

Structural half of S8 (aiolabs/satmachineadmin#22). Listener now
accepts BOTH inbound and outbound payments instead of filtering on
`is_in=True`; distribution gates the DCA leg on tx_type so the
liquidity-flow direction at the ATM drives behaviour, not the
Lightning protocol direction at the operator's wallet.

tasks.py:
 - Drop the `if not payment.is_in` pre-filter; keep `payment.success`.
 - Pair-name the two axes (`is_lightning_inbound`/`_outbound` for
   protocol vs `tx_type ∈ {cash_out, cash_in}` for business) per
   the naming-inversion memory.
 - Outbound payments need `extra.source == "bitspire"` before we
   touch them — without it we can't tell the operator paying their
   landlord from a cash-in settlement; skip silently.
 - Cross-axis sanity gate: refuse to process when protocol direction
   disagrees with business direction (cash_out must be inbound,
   cash_in must be outbound). Catches a buggy/malicious upstream
   stamping `type=cash_out` on an outbound payment.

distribution.py:
 - Gate `_pay_dca_distributions` on `tx_type == "cash_out"`. Cash-in
   liquidity stays in the operator's wallet — there's no LP share to
   distribute. Skipped leg is written as an audit row via
   `_record_skipped_leg` so the dashboard surfaces "DCA intentionally
   skipped" instead of a phantom missing leg.

Still pending in S8: the UI marker (cash_in tx_type chip in the
operator settlements table) and end-to-end test against a real
LNURL-withdraw redemption.

Tests: 75 passed (no regression vs prior green state; `test_router`
remains a pre-existing pytest-asyncio plugin issue).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

distribution.py

@@ -391,7 +391,28 @@ async def process_settlement(settlement_id: str) -> None:
     try:
         await _pay_super_fee(settlement, machine, super_config, errors)
         await _pay_operator_splits(settlement, machine, errors)
-        await _pay_dca_distributions(settlement, machine, errors)
+        # DCA distribution: applies to cash_out (LPs share the principal
+        # the customer paid into BTC). Does NOT apply to cash_in — that
+        # flow is liquidity coming IN to the operator's wallet, not
+        # going OUT to LPs. Skip with an audit row so the operator
+        # dashboard surfaces "DCA intentionally skipped for cash_in
+        # settlement" rather than displaying a phantom missing leg.
+        # See aiolabs/satmachineadmin#22 (S8 — wire cash-in path).
+        if settlement.tx_type == "cash_out":
+            await _pay_dca_distributions(settlement, machine, errors)
+        else:
+            await _record_skipped_leg(
+                settlement,
+                machine,
+                leg_type="dca",
+                amount_sats=settlement.principal_sats,
+                reason=(
+                    f"DCA distribution does not apply to tx_type="
+                    f"{settlement.tx_type!r}; principal stays in the "
+                    "operator's wallet as liquidity received from the "
+                    "cash-in customer."
+                ),
+            )
     except Exception as exc:  # last-resort guard
         logger.exception("distribution: unexpected error processing settlement")
         errors.append(f"unexpected: {exc}")

tasks.py

@@ -73,13 +73,41 @@ async def wait_for_paid_invoices() -> None:
 
 
 async def _handle_payment(payment: Payment) -> None:
-    if not payment.is_in or not payment.success:
+    if not payment.success:
         return
     machine = await get_active_machine_by_wallet_id(payment.wallet_id)
     if machine is None:
         return
     extra = payment.extra or {}
 
+    # Two axes, deliberately named in pairs to avoid the inversion trap
+    # documented at `~/.claude/projects/.../memory/feedback_naming_business_vs_protocol.md`:
+    #
+    #   - is_lightning_inbound / is_lightning_outbound: PROTOCOL direction
+    #     at the operator's wallet. `payment.is_in` from LNbits.
+    #   - tx_type ∈ {"cash_out", "cash_in"}: BUSINESS direction at the ATM.
+    #     Sourced from Payment.extra (canonical, stamped by bitSpire).
+    #
+    # Canonical mapping:
+    #   cash_out ↔ is_lightning_inbound  (customer pays ATM's invoice in BTC,
+    #                                     operator wallet receives sats)
+    #   cash_in  ↔ is_lightning_outbound (customer redeems ATM's LNURL-
+    #                                     withdraw, operator wallet sends sats)
+    #
+    # Process BOTH directions; reject mismatches at the discriminator gate.
+    is_lightning_inbound = payment.is_in
+    is_lightning_outbound = not payment.is_in
+
+    # Outbound payments from the operator's wallet need an extra
+    # discriminator before we touch them. An operator may legitimately
+    # send sats for non-ATM reasons (manual send, different extension,
+    # etc.). Without `source=bitspire` on Payment.extra we can't tell
+    # the operator paying their landlord from a cash-in settlement —
+    # skip silently. (For cash-out / inbound payments we already gate
+    # on machine-owned wallet via `get_active_machine_by_wallet_id`.)
+    if is_lightning_outbound and extra.get("source") != "bitspire":
+        return
+
     # 1) Attribution FIRST — uses only `extra.nostr_sender_pubkey` (no parse
     # needed). If this fails, every subsequent field on `extra` is
     # attacker-controlled and untrustworthy — record a minimal rejected
@@ -112,6 +140,26 @@ async def _handle_payment(payment: Payment) -> None:
         await _record_rejected(payment, machine, exc)
         return
 
+    # Cross-axis sanity: protocol direction must agree with business
+    # direction per the canonical mapping above. A mismatch means
+    # something upstream is confused — refuse to process. Concrete
+    # symptom this catches: an attacker (or a buggy extension) stamps
+    # `source=bitspire, type=cash_out` on an outbound payment from the
+    # operator's wallet to attempt a fake "we just received sats" row.
+    expected_inbound = data.tx_type == "cash_out"
+    if is_lightning_inbound != expected_inbound:
+        await _record_rejected(
+            payment,
+            machine,
+            SettlementInvariantError(
+                f"direction mismatch: payment.is_in={is_lightning_inbound} "
+                f"but tx_type={data.tx_type!r}. Expected cash_out ↔ inbound, "
+                "cash_in ↔ outbound."
+            ),
+        )
+        return
+    del is_lightning_outbound  # only used for the discriminator above
+
     # Stamp the originating Nostr event id (the kind-21000 create_invoice
     # RPC) onto the row for post-hoc forensics — an auditor can trace
     # settlement → RPC event → signing key without trusting our DB.