feat(v2): settlement distribution — three leg groups, super-fee write (P2)

After a settlement lands (P1a), this commit pays out the three leg
groups via LNbits internal transfers (create_invoice + pay_invoice with
internal=True). Wired synchronously from the invoice listener — latency
is one bitSpire-tx wide. process_settlement is idempotent (status guard)
so retries are safe.

distribution.py — three leg groups, in order:

  1. super_fee leg:
       platform_fee_sats → super_fee_wallet_id (if set)
       skip + warn if super fee % > 0 but wallet not configured
  2. operator_split legs:
       operator_fee_sats sliced per the operator's commission_splits
       ruleset (per-machine override or operator default)
       skip + warn if operator has no ruleset configured
  3. dca legs:
       net_sats distributed proportionally to active flow-mode LPs at
       this machine, each capped at the LP's remaining-fiat-balance-
       in-sats (preserves the v1 sync-mismatch fix from PR #2)
       skip if exchange_rate=0 (fallback path with missing rate)

Every leg lands a dca_payments row with the leg_type discriminator and
inherits Payment.tag "satmachine:{machine_npub}" so LNbits payment-
history filters work natively across machines + operators.

Atomicity model: LN payments cannot be rolled back. Each leg is
attempted independently; success/fail recorded on the dca_payments row.
The settlement is marked 'processed' only when every leg completed; any
failure marks 'errored' with a concatenated message but leaves successful
legs in place. Sats that don't pay out (failed legs, missing super
wallet, no commission ruleset, no LP coverage) remain in the machine's
wallet — visible to the operator on the dashboard.

calculations.py — extracted two pure helpers:

  split_two_stage_commission(commission_sats, super_fee_pct)
    Stage-1: super takes super_fee_pct (rounded); operator absorbs the
    rounding remainder so platform + operator == commission_sats exactly.

  allocate_operator_split_legs(operator_fee_sats, leg_pcts)
    Stage-2: distributes the remainder across N legs per pct rules. Last
    leg absorbs the rounding remainder so sum(legs) == operator_fee_sats.

50 new tests cover the plan's verification scenario:
  100 sats commission, super=30%, operator splits 50/30/20
  → super 30, operator 35/21/14. Sum 100 ✓
plus all the edge cases the plan called out (super=0, super=100,
single-leg, zero-fee, parametrised invariant on sums).

views_api.py adds the super-only platform-fee write endpoint:
  PUT /api/v1/dca/super-config  (check_super_user)

This is the only super-only endpoint in v2 — sets super_fee_pct and the
destination wallet for collecting the fee.

72/72 tests pass (22 calculation + 50 two-stage-split). 13 routes
registered against LNbits 1.4 (nostr-transport).

Refs: aiolabs/satmachineadmin#9

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

tests/test_two_stage_split.py

@@ -0,0 +1,144 @@
+"""
+Tests for the v2 two-stage commission split (super first, operator remainder).
+
+The plan calls out a verification scenario explicitly:
+  super_fee_pct=30%, operator split 50/30/20 on a 100-sat commission
+  → super_wallet gets 30, operator_self gets 35, employee 21, maint 14.
+
+Also covers the edge cases: super_fee_pct=0 (no super), super_fee_pct=1.0
+(everything to super), single-leg operator ruleset, zero operator fee.
+"""
+
+import pytest
+
+from ..calculations import (
+    allocate_operator_split_legs,
+    split_two_stage_commission,
+)
+
+
+class TestSplitTwoStageCommission:
+    """Stage-1: super takes super_fee_pct of commission; operator gets rest."""
+
+    def test_plan_example_100sats_30pct(self):
+        platform, operator = split_two_stage_commission(100, 0.30)
+        assert platform == 30
+        assert operator == 70
+        assert platform + operator == 100
+
+    def test_realistic_7965sats_30pct(self):
+        # From the plan's 2000 GTQ → 266800 sats @ 3% commission example.
+        platform, operator = split_two_stage_commission(7965, 0.30)
+        assert platform == 2390  # round(7965 * 0.30) = 2389.5 → 2390
+        assert operator == 5575  # 7965 - 2390
+        assert platform + operator == 7965
+
+    def test_super_pct_zero_leaves_all_to_operator(self):
+        platform, operator = split_two_stage_commission(7965, 0.0)
+        assert platform == 0
+        assert operator == 7965
+
+    def test_super_pct_one_takes_everything(self):
+        platform, operator = split_two_stage_commission(7965, 1.0)
+        assert platform == 7965
+        assert operator == 0
+
+    def test_zero_commission(self):
+        platform, operator = split_two_stage_commission(0, 0.30)
+        assert platform == 0
+        assert operator == 0
+
+    def test_negative_commission_clamps_to_zero(self):
+        # Defensive: should never happen, but verify we don't go negative.
+        platform, operator = split_two_stage_commission(-100, 0.30)
+        assert platform == 0
+        assert operator == 0
+
+    @pytest.mark.parametrize("commission_sats", [1, 7, 100, 7965, 1_000_000])
+    @pytest.mark.parametrize("super_pct", [0.0, 0.1, 0.30, 0.5, 0.777, 1.0])
+    def test_invariant_sum_equals_commission(self, commission_sats, super_pct):
+        platform, operator = split_two_stage_commission(commission_sats, super_pct)
+        assert platform + operator == commission_sats
+        assert 0 <= platform <= commission_sats
+        assert 0 <= operator <= commission_sats
+
+
+class TestAllocateOperatorSplitLegs:
+    """Stage-2: operator's remainder split across N leg wallets per pct rules."""
+
+    def test_plan_example_50_30_20_on_70(self):
+        amounts = allocate_operator_split_legs(70, [0.5, 0.3, 0.2])
+        assert amounts == [35, 21, 14]
+        assert sum(amounts) == 70
+
+    def test_realistic_50_30_20_on_5575(self):
+        amounts = allocate_operator_split_legs(5575, [0.5, 0.3, 0.2])
+        # 50%: round(2787.5) = 2788; 30%: round(1672.5) = 1672; last absorbs
+        # remainder: 5575 - 2788 - 1672 = 1115.
+        # Note: round() uses banker's rounding so 2787.5 → 2788 actually
+        # because 2788 is even. Confirm by total invariant.
+        assert sum(amounts) == 5575
+        assert len(amounts) == 3
+
+    def test_single_leg_full_remainder(self):
+        amounts = allocate_operator_split_legs(100, [1.0])
+        assert amounts == [100]
+
+    def test_zero_operator_fee_zeros_all_legs(self):
+        amounts = allocate_operator_split_legs(0, [0.5, 0.5])
+        assert amounts == [0, 0]
+
+    def test_empty_legs_list_returns_empty(self):
+        amounts = allocate_operator_split_legs(100, [])
+        assert amounts == []
+
+    def test_last_leg_absorbs_rounding_remainder(self):
+        # 100 / 3 ≈ 33.33 each; rounding makes the first two 33 and last 34.
+        amounts = allocate_operator_split_legs(100, [1 / 3, 1 / 3, 1 / 3])
+        assert sum(amounts) == 100
+        assert amounts[0] == round(100 / 3)  # 33
+        assert amounts[1] == round(100 / 3)  # 33
+        # Last leg absorbs the rounding (34, not 33) so total == 100.
+        assert amounts[2] == 100 - amounts[0] - amounts[1]
+
+    @pytest.mark.parametrize(
+        "operator_fee,pcts",
+        [
+            (1, [0.5, 0.5]),
+            (7, [0.5, 0.3, 0.2]),
+            (100, [0.5, 0.5]),
+            (5575, [0.5, 0.3, 0.2]),
+            (1_000_000, [0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1]),
+        ],
+    )
+    def test_invariant_sum_equals_operator_fee(self, operator_fee, pcts):
+        amounts = allocate_operator_split_legs(operator_fee, pcts)
+        assert sum(amounts) == operator_fee
+        assert all(a >= 0 for a in amounts)
+
+
+class TestEndToEndScenarios:
+    """The full two-stage split — super then operator legs — composed."""
+
+    def test_plan_example_full(self):
+        # 100 sats commission, super=30%, operator splits 50/30/20.
+        platform, operator = split_two_stage_commission(100, 0.30)
+        legs = allocate_operator_split_legs(operator, [0.5, 0.3, 0.2])
+        assert platform == 30
+        assert legs == [35, 21, 14]
+        assert platform + sum(legs) == 100
+
+    def test_super_pct_zero_full_pipeline(self):
+        platform, operator = split_two_stage_commission(7965, 0.0)
+        legs = allocate_operator_split_legs(operator, [1.0])
+        assert platform == 0
+        assert legs == [7965]
+        assert platform + sum(legs) == 7965
+
+    def test_super_pct_one_full_pipeline(self):
+        platform, operator = split_two_stage_commission(7965, 1.0)
+        legs = allocate_operator_split_legs(operator, [0.5, 0.5])
+        assert platform == 7965
+        # Operator has zero to distribute; both legs get zero.
+        assert legs == [0, 0]
+        assert platform + sum(legs) == 7965