feat(v2): partial-dispense + operator notes on settlements (P3d)

Closes the v1 feature request satmachineadmin#3 (partial transaction
processing) and adds operator-authored audit notes on settlements.

Schema (m006_add_settlement_notes):
  ALTER TABLE dca_settlements ADD COLUMN notes TEXT

The notes column is append-only (prepend with timestamp, never edit in
place). Stores both system-generated audit memos (partial-dispense
recompute provenance) and operator-authored free-form notes (cash-
drawer reconciliation context, off-LN refund records, etc.).

Partial-dispense endpoint:
  POST /api/v1/dca/settlements/{id}/partial-dispense
  body: PartialDispenseData {dispensed_fraction OR dispensed_sats, notes}

Recompute path (in distribution.apply_partial_dispense_and_redistribute):
  1. Refuse if any leg has status='completed' (Lightning can't claw back)
  2. Resolve new_gross from dispensed_fraction or dispensed_sats
  3. Linear-scale net/commission/fiat — preserves the original commission
     ratio exactly; only rounding may drift by 1 sat
  4. Re-stage-1 split using the CURRENT super_fee_pct (super may have
     changed the rate since the original landed)
  5. Build a memo capturing original values + reason + new values
  6. Void pending/failed legs (status → 'voided')
  7. Overwrite the settlement's monetary fields + prepend memo to notes
  8. Reset status to 'pending' → process_settlement re-runs distribution

Operator notes endpoint:
  POST /api/v1/dca/settlements/{id}/notes
  body: AppendSettlementNoteData {note}

Each operator note is timestamped (UTC) and tagged with the author's
user_id so the audit trail is accountable. Non-empty, max 2000 chars.

72/72 tests still pass. 30 routes total. The full-directory ruff number
ballooned to ~500 because it includes legacy transaction_processor.py
(orphaned, not imported anywhere) and other v1 cruft on the branch.
Files I actively maintain are clean.

Note: a richer queryable audit history (filter by author / time range /
action type / etc.) is being tracked as a separate future-work issue.
The notes-column approach here is the v1 audit story; the dedicated
history table will be additive.

Refs: aiolabs/satmachineadmin#9, closes #3 (in spirit, marked
once verified end-to-end)

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

distribution.py

@@ -23,14 +23,20 @@
 
 from __future__ import annotations
 
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import List
 
 from lnbits.core.services import create_invoice, pay_invoice
 from loguru import logger
 
-from .calculations import allocate_operator_split_legs, calculate_distribution
+from .calculations import (
+    allocate_operator_split_legs,
+    calculate_distribution,
+    split_two_stage_commission,
+)
 from .crud import (
+    apply_partial_dispense,
+    count_completed_legs_for_settlement,
     create_dca_payment,
     get_client_balance_summary,
     get_effective_commission_splits,
@@ -40,12 +46,14 @@ from .crud import (
     get_super_config,
     mark_settlement_status,
     update_payment_status,
+    void_open_legs_for_settlement,
 )
 from .models import (
     CreateDcaPaymentData,
     DcaPayment,
     DcaSettlement,
     Machine,
+    PartialDispenseData,
     SuperConfig,
 )
 
@@ -56,6 +64,134 @@ def _payment_tag(machine: Machine) -> str:
     return f"{PAYMENT_TAG_PREFIX}:{machine.machine_npub}"
 
 
+def _resolve_partial_dispense_gross(
+    settlement: DcaSettlement, data: PartialDispenseData
+) -> int:
+    if data.dispensed_sats is not None:
+        new_gross = int(data.dispensed_sats)
+    elif data.dispensed_fraction is not None:
+        new_gross = round(settlement.gross_sats * float(data.dispensed_fraction))
+    else:
+        raise ValueError("provide one of dispensed_sats or dispensed_fraction")
+    if new_gross < 0:
+        raise ValueError("partial dispense cannot be negative")
+    if new_gross > settlement.gross_sats:
+        raise ValueError(
+            f"partial dispense ({new_gross} sats) cannot exceed the original "
+            f"gross ({settlement.gross_sats} sats)"
+        )
+    return new_gross
+
+
+def _build_partial_dispense_memo(
+    settlement: DcaSettlement,
+    data: PartialDispenseData,
+    *,
+    new_gross: int,
+    new_net: int,
+    new_commission: int,
+    new_platform: int,
+    new_operator: int,
+) -> str:
+    reason = (data.notes or "").strip() or "(no reason given)"
+    if data.dispensed_sats is not None:
+        adjust = f"dispensed_sats={data.dispensed_sats}"
+    else:
+        adjust = f"dispensed_fraction={data.dispensed_fraction}"
+    ts = datetime.now(timezone.utc).isoformat(timespec="seconds")
+    return (
+        f"[{ts}] partial dispense applied — {adjust}. "
+        f"Original gross={settlement.gross_sats} net={settlement.net_sats} "
+        f"commission={settlement.commission_sats} "
+        f"(super_fee={settlement.platform_fee_sats} "
+        f"operator_fee={settlement.operator_fee_sats}). "
+        f"New gross={new_gross} net={new_net} commission={new_commission} "
+        f"(super_fee={new_platform} operator_fee={new_operator}). "
+        f"Reason: {reason}"
+    )
+
+
+async def apply_partial_dispense_and_redistribute(
+    settlement_id: str, data: PartialDispenseData
+) -> DcaSettlement:
+    """Operator UX action — closes satmachineadmin#3.
+
+    When a bitSpire dispense fails mid-transaction (e.g., dispenser jam after
+    6 of 10 bills), the operator confirms the actual amount dispensed and we
+    re-allocate the split against that partial gross. Sat amounts scale
+    linearly, preserving the original commission ratio exactly; the two-stage
+    super/operator split is recomputed using the CURRENT super_fee_pct
+    (super may have changed the rate since the original landed).
+
+    Hard guard: refuses if any dca_payments leg has already completed.
+    Lightning payments can't be clawed back, so we won't try.
+
+    Side effects:
+      - Voids pending/failed legs (status → 'voided').
+      - Overwrites the settlement's monetary fields with the new totals.
+      - Appends a timestamped memo to settlement.notes capturing the
+        original values + operator's reason.
+      - Resets settlement.status to 'pending' and triggers process_settlement.
+    """
+    settlement = await get_settlement(settlement_id)
+    if settlement is None:
+        raise ValueError(f"settlement {settlement_id} not found")
+    if settlement.gross_sats <= 0:
+        raise ValueError("cannot partial-dispense a zero-gross settlement")
+    completed = await count_completed_legs_for_settlement(settlement_id)
+    if completed > 0:
+        raise ValueError(
+            f"cannot partial-dispense: {completed} leg(s) already completed "
+            "(Lightning payments can't be clawed back)"
+        )
+
+    new_gross = _resolve_partial_dispense_gross(settlement, data)
+    # Linear scale preserves the original commission ratio exactly.
+    scale = new_gross / settlement.gross_sats
+    new_commission = round(settlement.commission_sats * scale)
+    new_net = new_gross - new_commission
+    new_fiat = round(float(settlement.fiat_amount) * scale, 2)
+
+    # Re-stage-1 split using the CURRENT super_fee_pct.
+    super_config = await get_super_config()
+    super_fee_pct = float(super_config.super_fee_pct) if super_config else 0.0
+    new_platform, new_operator = split_two_stage_commission(
+        new_commission, super_fee_pct
+    )
+
+    memo = _build_partial_dispense_memo(
+        settlement,
+        data,
+        new_gross=new_gross,
+        new_net=new_net,
+        new_commission=new_commission,
+        new_platform=new_platform,
+        new_operator=new_operator,
+    )
+
+    await void_open_legs_for_settlement(settlement_id)
+    updated = await apply_partial_dispense(
+        settlement_id,
+        new_gross_sats=new_gross,
+        new_net_sats=new_net,
+        new_commission_sats=new_commission,
+        new_platform_fee_sats=new_platform,
+        new_operator_fee_sats=new_operator,
+        new_fiat_amount=new_fiat,
+        appended_note=memo,
+    )
+    if updated is None:
+        raise ValueError(f"settlement {settlement_id} disappeared mid-update")
+
+    logger.info(
+        f"distribution: partial-dispense applied to settlement "
+        f"{settlement_id} — re-running distribution"
+    )
+    await process_settlement(settlement_id)
+    after = await get_settlement(settlement_id)
+    return after if after is not None else updated
+
+
 async def process_settlement(settlement_id: str) -> None:
     """Process a pending settlement end-to-end. Safe to invoke multiple
     times — the status='processed' guard skips already-processed rows."""
@@ -155,7 +291,7 @@ async def _pay_operator_splits(
         settlement.operator_fee_sats,
         [float(leg.pct) for leg in splits],
     )
-    for idx, (leg, amount) in enumerate(zip(splits, leg_amounts)):
+    for idx, (leg, amount) in enumerate(zip(splits, leg_amounts, strict=True)):
         if amount <= 0:
             continue
         label = leg.label or f"split-{idx + 1}"