docs(security-pathway): flag kind:21001 CLINK collision + rotation

S3 settlement-receipt kind was provisionally 21001, but that kind is
claimed by CLINK (Offers). Replace the speculative kind text in the
cash-out diagram and the S3 row with an explicit DO-NOT-USE alert
citing the 2026-06-02 collision, the aiolabs/satmachineadmin#44
rotation tracker, and the 22000-22099 target band.

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

bitspire.py

@@ -17,7 +17,10 @@ from __future__ import annotations
 import json
 from typing import Any, Optional
 
-from .models import CreateDcaSettlementData, Machine
+from loguru import logger
+
+from .calculations import split_principal_based
+from .models import CreateDcaSettlementData, Machine, SuperConfig
 
 # Sentinel value bitSpire sets in Payment.extra.source so we know an inbound
 # payment originated from an ATM cash-out and not some other extension or
@@ -219,23 +222,30 @@ def parse_settlement(
     payment_hash: str,
     wire_sats: int,
     extra: dict,
-    super_fee_fraction: float,
+    super_config: SuperConfig,
 ) -> CreateDcaSettlementData:
     """Build a CreateDcaSettlementData for an inbound payment landing on
     `machine`'s wallet.
 
+    Splits the fee on a principal-based, direction-aware model
+    (aiolabs/satmachineadmin#37,#38):
+
+      platform_fee_sats = round(principal_sats * super_cash_{type}_fee_fraction)
+      operator_fee_sats = round(principal_sats * operator_cash_{type}_fee_fraction)
+
+    where the directional super fraction comes from `super_config` and
+    the operator fraction comes from `machine`. The bitspire-reported
+    `fee_sats` field is preserved on the settlement as the customer's
+    actual paid total, but is NOT used as input to the split.
+
     Requires bitSpire's canonical Payment.extra stamp (source="bitspire"
     plus the absolute sat amounts) per aiolabs/lamassu-next#44. Raises
     `SettlementMetadataError` on missing/partial stamp — caller records
     the settlement as 'rejected' for upstream investigation. Raises
     `SettlementInvariantError` if the stamped values violate the
     canonical sat-amount invariants (range + sum, see
-    `_assert_sat_invariants`).
+    `_assert_sat_invariants`) or `tx_type` is unknown.
     """
-    if not (0.0 <= super_fee_fraction <= 1.0):
-        raise SettlementInvariantError(
-            f"super_fee_fraction must be in [0, 1], got {super_fee_fraction}"
-        )
     if not is_bitspire_payment(extra):
         raise SettlementMetadataError(
             f"Payment.extra missing `source: \"bitspire\"` marker on machine "
@@ -253,8 +263,39 @@ def parse_settlement(
             f"(lamassu-next#44) requires both. Investigate the ATM "
             f"firmware on machine {machine.machine_npub[:12]}..."
         )
-    platform_fee_sats = round(fee_sats * super_fee_fraction)
-    operator_fee_sats = fee_sats - platform_fee_sats
+    tx_type = _coerce_str(extra.get("type")) or "cash_out"
+    if tx_type == "cash_in":
+        super_frac = float(super_config.super_cash_in_fee_fraction)
+        operator_frac = float(machine.operator_cash_in_fee_fraction)
+    elif tx_type == "cash_out":
+        super_frac = float(super_config.super_cash_out_fee_fraction)
+        operator_frac = float(machine.operator_cash_out_fee_fraction)
+    else:
+        raise SettlementInvariantError(
+            f"unknown tx_type={tx_type!r}; expected 'cash_in' or 'cash_out'"
+        )
+    platform_fee_sats, operator_fee_sats = split_principal_based(
+        principal_sats, super_frac, operator_frac
+    )
+    # Phase-1 observability per aiolabs/satmachineadmin#38 + coord-log
+    # §2026-06-01T07:00Z (option A locked): compare bitspire's reported
+    # fee_sats against satmachineadmin's recompute, log on out-of-
+    # tolerance drift, record the delta unconditionally for triage.
+    # Phase 2 (settlement-reject) lands after observability data.
+    fee_mismatch_sats = fee_sats - (platform_fee_sats + operator_fee_sats)
+    tolerance = max(1, int(principal_sats * 0.001))
+    if abs(fee_mismatch_sats) > tolerance:
+        logger.warning(
+            f"bitspire fee mismatch on payment {payment_hash[:12]}...: "
+            f"bitspire_fee_sats={fee_sats} expected={platform_fee_sats + operator_fee_sats} "
+            f"delta={fee_mismatch_sats} tolerance={tolerance} "
+            f"principal={principal_sats} super_frac={super_frac:.4f} "
+            f"operator_frac={operator_frac:.4f} tx_type={tx_type} "
+            f"machine={machine.machine_npub[:12]}... — "
+            "Phase 1 observability only, no behavior change. Pre-Layer-3 "
+            "(lamassu-next#57) the ATM still hardcodes fee fractions, so "
+            "large deltas here are expected until that ships."
+        )
     exchange_rate = _coerce_float(extra.get("exchange_rate"))
     if exchange_rate is None or exchange_rate <= 0:
         # Without exchange rate we can't compute fiat. Use 1.0 as a stand-in
@@ -268,7 +309,6 @@ def parse_settlement(
     # in BTC today, but the cash side has its own ground truth).
     fiat_amount = _coerce_float(extra.get("fiat_amount")) or 0.0
     fiat_code = _coerce_str(extra.get("currency")) or machine.fiat_code
-    tx_type = _coerce_str(extra.get("type")) or "cash_out"
     data = CreateDcaSettlementData(
         machine_id=machine.id,
         payment_hash=payment_hash,
@@ -282,6 +322,7 @@ def parse_settlement(
         fee_sats=fee_sats,
         platform_fee_sats=platform_fee_sats,
         operator_fee_sats=operator_fee_sats,
+        fee_mismatch_sats=fee_mismatch_sats,
         tx_type=tx_type,
         bills_json=_json_dumps(extra.get("bills")),
         cassettes_json=_json_dumps(extra.get("cassettes")),

calculations.py

@@ -106,36 +106,49 @@ def calculate_distribution(
     return distributions
 
 
-def split_two_stage_commission(
-    fee_sats: int, super_fee_fraction: float
+def split_principal_based(
+    principal_sats: int,
+    super_frac: float,
+    operator_frac: float,
 ) -> Tuple[int, int]:
-    """Stage-1 of the v2 commission split: super takes `super_fee_fraction`
-    of the total fee; the remainder is what the operator's own ruleset
-    acts on.
+    """Compute platform + operator fee shares as independent fractions of
+    `principal_sats`. Both shares are derived from the customer's
+    principal (the canonical source of truth), NOT back-derived from
+    `fee_sats`.
 
-    Returns (platform_fee_sats, operator_fee_sats). Platform is rounded;
-    operator absorbs the rounding remainder so platform_fee + operator_fee
-    == fee_sats exactly.
+    Returns (platform_fee_sats, operator_fee_sats). Both are rounded
+    independently; rounding remainders do NOT compound — the customer
+    pays whatever bitspire collected, and any drift between (super +
+    operator) and the bitspire-reported `fee_sats` surfaces via
+    `dca_settlements.fee_mismatch_sats`.
 
     Examples:
-        >>> split_two_stage_commission(100, 0.30)
-        (30, 70)
-        >>> split_two_stage_commission(7965, 0.30)
-        (2390, 5575)
-        >>> split_two_stage_commission(100, 0.0)
-        (0, 100)
-        >>> split_two_stage_commission(100, 1.0)
-        (100, 0)
+        >>> split_principal_based(100_000, 0.03, 0.05)
+        (3000, 5000)
+        >>> split_principal_based(266_800, 0.03, 0.0)
+        (8004, 0)
+        >>> split_principal_based(100_000, 0.0, 0.0)
+        (0, 0)
+        >>> split_principal_based(100_000, 0.15, 0.0)
+        (15000, 0)
+
+    The pre-#38 bug this corrects: the old math interpreted the super
+    fee as `fraction_of_fee` rather than `fraction_of_principal`. On a
+    100_000-sat principal with an 8% total bitspire fee (= 8_000 sats
+    fee_sats) and super_fraction=0.03, the bug paid the super
+    `round(8_000 * 0.03) = 240` sats — ~13× below the intended
+    `100_000 * 0.03 = 3_000` sats per-settlement. Repeated on every
+    cash-out since the bitspire wire-shape landed. See
+    aiolabs/satmachineadmin#37 (parent) + #38 (this layer).
     """
-    if not (0.0 <= super_fee_fraction <= 1.0):
-        raise ValueError(
-            f"super_fee_fraction must be in [0, 1], got {super_fee_fraction}"
-        )
-    if fee_sats <= 0:
+    if not (0.0 <= super_frac <= 1.0):
+        raise ValueError(f"super_frac must be in [0, 1], got {super_frac}")
+    if not (0.0 <= operator_frac <= 1.0):
+        raise ValueError(f"operator_frac must be in [0, 1], got {operator_frac}")
+    if principal_sats <= 0:
         return 0, 0
-    platform = round(fee_sats * super_fee_fraction)
-    platform = max(0, min(platform, fee_sats))
-    operator = fee_sats - platform
+    platform = max(0, round(principal_sats * super_frac))
+    operator = max(0, round(principal_sats * operator_frac))
     return platform, operator
 
 

cassette_transport.py

@@ -41,57 +41,61 @@ centralises the canonicalisation via lnbits.utils.nostr.normalize_public_key.
 from __future__ import annotations
 
 import json
-import time
 
-from lnbits.core.crud.users import get_account
 from lnbits.core.services.nip46_bunker_client import (
     NsecBunkerRpcError,
     NsecBunkerTimeoutError,
 )
-from lnbits.core.signers import resolve_signer
 from lnbits.core.signers.base import (
     NostrSigner,
-    SignerError,
     SignerUnavailableError,
 )
 from lnbits.utils.nostr import normalize_public_key
-from loguru import logger
 
 from .models import Machine, PublishCassettesPayload
 from .nip44 import Nip44Error
-from .nip44 import decrypt_from as _nip44_local_decrypt
-from .nip44 import encrypt_for as _nip44_local_encrypt
+from .nostr_publish import (
+    NostrPublishError,
+    OperatorIdentityMissing,  # re-export for callers that catch this
+    RelayUnavailable,  # re-export
+    SignerUnavailable,  # re-export
+    nip44_decrypt_via_signer,
+    publish_encrypted_kind_30078,
+)
+
+# Re-exported so external callers (views_api etc.) can keep importing
+# from cassette_transport without breakage. Same for the public
+# constants below.
+__all__ = [
+    "CassetteTransportError",
+    "CassetteEventDecodeError",
+    "CassetteEventTransientError",
+    "OperatorIdentityMissing",
+    "SignerUnavailable",
+    "RelayUnavailable",
+    "build_state_d_tags_for_machines",
+    "decrypt_and_parse_state_event",
+    "publish_to_atm",
+]
 
-_KIND_NIP78 = 30078
 _D_TAG_CONFIG_PREFIX = "bitspire-cassettes:"  # operator → ATM
 _D_TAG_STATE_PREFIX = "bitspire-cassettes-state:"  # ATM → operator
 
 
 # =============================================================================
-# Errors
+# Errors — cassette-specific subclasses of the generic NostrPublishError
 # =============================================================================
 
 
-class CassetteTransportError(Exception):
-    """Generic transport-layer error. Subclasses distinguish failure modes
-    so the API can surface meaningful HTTP statuses + the consumer task
-    can log + skip without crashing."""
+class CassetteTransportError(NostrPublishError):
+    """Generic cassette-transport error. Subclasses distinguish failure
+    modes so the API can surface meaningful HTTP statuses + the consumer
+    task can log + skip without crashing.
 
-
-class OperatorIdentityMissing(CassetteTransportError):
-    """Operator account has no Nostr pubkey on file, or no signer is
-    available (pre-bunker rollout — operator hasn't onboarded via
-    Nostr-login)."""
-
-
-class SignerUnavailable(CassetteTransportError):
-    """Resolved signer can't sign server-side (client-side-only signer,
-    or transient bunker unreachability post-lnbits#18). Publish skipped."""
-
-
-class RelayUnavailable(CassetteTransportError):
-    """nostrclient extension isn't installed or its relay manager isn't
-    reachable. Treated as soft-fail; publish skipped + logged."""
+    Bridges back-compat with pre-extraction callers that catch this
+    class — now equivalent to NostrPublishError plus the two consumer-
+    side decode/transient distinctions below.
+    """
 
 
 class CassetteEventDecodeError(CassetteTransportError):
@@ -141,139 +145,11 @@ def build_state_d_tags_for_machines(machines: list[Machine]) -> list[str]:
     return [_state_d_tag(_atm_hex_pubkey(m)) for m in machines]
 
 
-# =============================================================================
-# Sign-as-operator — hybrid path (resolve_signer post #17, prvkey fallback)
-# =============================================================================
-
-
-async def _resolve_operator_signer(operator_user_id: str):
-    """Fetch the operator's account + resolve to a NostrSigner.
-
-    Single source of truth for "give me the signer for this operator,
-    or raise an operator-facing error if we can't." Returns
-    `(account, signer)` so callers that need both (publish path needs
-    `account.pubkey` for the event author and the signer for both
-    encrypt + sign) don't double-fetch.
-
-    Raises:
-      - OperatorIdentityMissing — no account, or no pubkey on file
-      - SignerUnavailable — signer resolve failed, or signer can't sign
-        server-side (ClientSideOnly)
-    """
-    account = await get_account(operator_user_id)
-    if account is None or not account.pubkey:
-        raise OperatorIdentityMissing(
-            f"operator {operator_user_id[:8]}... has no Nostr pubkey on file. "
-            "Onboard via the LNbits Nostr-login flow to publish cassette "
-            "config to your ATMs."
-        )
-    try:
-        signer = resolve_signer(account)
-    except SignerError as exc:
-        raise SignerUnavailable(
-            f"signer resolve failed for operator {operator_user_id[:8]}...: " f"{exc}"
-        ) from exc
-    if not signer.can_sign():
-        raise SignerUnavailable(
-            f"operator {operator_user_id[:8]}... has a client-side-only "
-            "signer; server can't sign or NIP-44-encrypt on their behalf. "
-            "Operator must hold their nsec via a NIP-46 bunker (lnbits#18) "
-            "or migrate to a server-signing account."
-        )
-    return account, signer
-
-
-async def _sign_as_operator(operator_user_id: str, event: dict) -> dict | None:
-    """Sign `event` using the operator's signer (LocalSigner or
-    RemoteBunkerSigner). Mutates `event` to add `created_at` (now),
-    `pubkey`, `id`, and `sig`.
-
-    Raises typed CassetteTransportError subclasses on hard failure
-    (the publish endpoint maps these to HTTP statuses); never returns
-    None on the publish path.
-    """
-    _account, signer = await _resolve_operator_signer(operator_user_id)
-    # created_at is part of the BIP-340 event-id hash; set before signing.
-    event["created_at"] = int(time.time())
-    try:
-        return await signer.sign_event(event)
-    except SignerUnavailableError as exc:
-        raise SignerUnavailable(
-            f"signer unavailable for operator {operator_user_id[:8]}...: " f"{exc}"
-        ) from exc
-
-
-async def _nip44_encrypt_via_signer(
-    account, signer: NostrSigner, plaintext: str, peer_pubkey_hex: str
-) -> str:
-    """NIP-44 v2 encrypt via the signer abstraction, with a transitional
-    fallback to direct-prvkey for LocalSigner accounts.
-
-    The bunker (RemoteBunkerSigner) implements `nip44_encrypt` natively —
-    the operator's nsec never leaves the bunker process. LocalSigner's
-    `nip44_encrypt` stub explicitly raises SignerUnavailableError
-    ("LocalSigner does not implement nip44_encrypt") per the
-    post-PR-#38 ABC — the spec is "migrate to bunker." For the
-    transitional window where some operators are still on LocalSigner
-    + their `account.prvkey` is intact, we catch that signal and use
-    our hand-rolled NIP-44 v2 impl against the stored prvkey. Same
-    wire output either way.
-
-    Removed once every operator account on this instance is bunker-
-    backed (S7 fully landed). At that point this helper collapses to
-    `return await signer.nip44_encrypt(plaintext, peer_pubkey_hex)`.
-    """
-    try:
-        return await signer.nip44_encrypt(plaintext, peer_pubkey_hex)
-    except SignerUnavailableError:
-        if account.signer_type == "LocalSigner" and account.prvkey:
-            return _nip44_local_encrypt(plaintext, account.prvkey, peer_pubkey_hex)
-        # ClientSideOnly, or RemoteBunkerSigner with bunker comms failure
-        # at config time — re-raise without wrapping; caller maps it.
-        raise
-
-
-async def _nip44_decrypt_via_signer(
-    account, signer: NostrSigner, ciphertext: str, peer_pubkey_hex: str
-) -> str:
-    """Decrypt mirror of `_nip44_encrypt_via_signer`. Same LocalSigner
-    transitional fallback."""
-    try:
-        return await signer.nip44_decrypt(ciphertext, peer_pubkey_hex)
-    except SignerUnavailableError:
-        if account.signer_type == "LocalSigner" and account.prvkey:
-            return _nip44_local_decrypt(ciphertext, account.prvkey, peer_pubkey_hex)
-        raise
-
-
 # =============================================================================
 # Publish — operator → ATM (the satmachineadmin API path)
 # =============================================================================
 
 
-async def _publish_signed_event(signed_event: dict) -> None:
-    """Send a signed Nostr event to all relays via the nostrclient
-    extension's singleton RelayManager.
-
-    Lazy import + typed-error so the API can surface "your LNbits doesn't
-    have nostrclient installed" as a 503 rather than a 500. Pattern
-    matches the cross-extension import guards in
-    `lnbits.core.services.users` (nostrmarket / nostrrelay).
-    """
-    try:
-        from nostrclient.router import (  # type: ignore[import-not-found]
-            nostr_client,
-        )
-    except ImportError as exc:
-        raise RelayUnavailable(
-            "nostrclient extension is not installed; cassette config "
-            "publish requires it. Install + activate the nostrclient "
-            "extension on this LNbits instance."
-        ) from exc
-    msg = json.dumps(["EVENT", signed_event])
-    nostr_client.relay_manager.publish_message(msg)
-
-
 async def publish_to_atm(
     machine: Machine,
     payload: PublishCassettesPayload,
@@ -283,63 +159,20 @@ async def publish_to_atm(
     from the operator to the target ATM.
 
     Returns the signed event dict on success (caller may log event.id for
-    audit). Raises CassetteTransportError subclasses on hard failures:
-      - OperatorIdentityMissing → 400: operator hasn't onboarded
-      - SignerUnavailable → 503: signer offline / client-side-only / bunker
-        timeout at the encrypt or sign step
-      - RelayUnavailable → 503: nostrclient not installed
-      - CassetteTransportError → 500: anything else
+    audit). Raises NostrPublishError subclasses (re-exported here as
+    CassetteTransportError, OperatorIdentityMissing, SignerUnavailable,
+    RelayUnavailable) on hard failures.
     """
     atm_pubkey_hex = _atm_hex_pubkey(machine)
-
-    # Single fetch + resolve — same signer is used for both encrypt and sign.
-    account, signer = await _resolve_operator_signer(operator_user_id)
-
-    # NIP-44 v2 encrypt the wire payload. Bunker round-trip on
-    # RemoteBunkerSigner; direct prvkey on LocalSigner (transitional).
-    plaintext = json.dumps(payload.to_wire_dict(), separators=(",", ":"))
-    try:
-        content = await _nip44_encrypt_via_signer(
-            account, signer, plaintext, atm_pubkey_hex
-        )
-    except NsecBunkerTimeoutError as exc:
-        raise SignerUnavailable(
-            f"bunker unreachable while encrypting cassette config for "
-            f"operator {operator_user_id[:8]}...: {exc}"
-        ) from exc
-    except NsecBunkerRpcError as exc:
-        raise SignerUnavailable(
-            f"bunker rejected nip44_encrypt for operator "
-            f"{operator_user_id[:8]}... (policy / MAC / config issue): "
-            f"{exc}"
-        ) from exc
-    except SignerUnavailableError as exc:
-        raise SignerUnavailable(
-            f"signer cannot nip44-encrypt for operator "
-            f"{operator_user_id[:8]}...: {exc}"
-        ) from exc
-
-    event: dict = {
-        "kind": _KIND_NIP78,
-        "tags": [
-            ["d", _config_d_tag(atm_pubkey_hex)],
-            ["p", atm_pubkey_hex],
-        ],
-        "content": content,
-        # created_at is set inside _sign_as_operator before signing.
-    }
-    signed = await _sign_as_operator(operator_user_id, event)
-    if signed is None:
-        raise CassetteTransportError(
-            "sign_as_operator returned None unexpectedly — soft-fail path "
-            "shouldn't be reachable on a publish-initiated flow"
-        )
-
-    await _publish_signed_event(signed)
-    logger.info(
-        f"satmachineadmin: published kind-30078 cassette config to ATM "
-        f"{atm_pubkey_hex[:12]}... (event_id={signed['id'][:12]}..., "
-        f"machine_id={machine.id}, positions={sorted(payload.positions.keys())})"
+    signed = await publish_encrypted_kind_30078(
+        operator_user_id=operator_user_id,
+        recipient_pubkey_hex=atm_pubkey_hex,
+        d_tag=_config_d_tag(atm_pubkey_hex),
+        payload=payload.to_wire_dict(),
+        log_context=(
+            f"cassette config (machine={machine.id}, "
+            f"positions={sorted(payload.positions.keys())})"
+        ),
     )
     return signed
 
@@ -384,7 +217,7 @@ async def decrypt_and_parse_state_event(
         )
 
     try:
-        plaintext = await _nip44_decrypt_via_signer(
+        plaintext = await nip44_decrypt_via_signer(
             account, signer, content, sender_pubkey
         )
     except NsecBunkerTimeoutError as exc:

crud.py

@@ -80,9 +80,13 @@ async def create_machine(operator_user_id: str, data: CreateMachineData) -> Mach
         """
         INSERT INTO satoshimachine.dca_machines
         (id, operator_user_id, machine_npub, wallet_id, name, location,
-         fiat_code, is_active, created_at, updated_at)
+         fiat_code, is_active,
+         operator_cash_in_fee_fraction, operator_cash_out_fee_fraction,
+         created_at, updated_at)
         VALUES (:id, :operator_user_id, :machine_npub, :wallet_id, :name,
-                :location, :fiat_code, :is_active, :created_at, :updated_at)
+                :location, :fiat_code, :is_active,
+                :operator_cash_in_fee_fraction, :operator_cash_out_fee_fraction,
+                :created_at, :updated_at)
         """,
         {
             "id": machine_id,
@@ -93,6 +97,8 @@ async def create_machine(operator_user_id: str, data: CreateMachineData) -> Mach
             "location": data.location,
             "fiat_code": data.fiat_code,
             "is_active": True,
+            "operator_cash_in_fee_fraction": data.operator_cash_in_fee_fraction,
+            "operator_cash_out_fee_fraction": data.operator_cash_out_fee_fraction,
             "created_at": now,
             "updated_at": now,
         },
@@ -595,13 +601,13 @@ async def create_settlement_idempotent(
         INSERT INTO satoshimachine.dca_settlements
         (id, machine_id, payment_hash, bitspire_event_id, bitspire_txid,
          wire_sats, fiat_amount, fiat_code, exchange_rate, principal_sats,
-         fee_sats, platform_fee_sats, operator_fee_sats,
+         fee_sats, platform_fee_sats, operator_fee_sats, fee_mismatch_sats,
          tx_type, bills_json, cassettes_json,
          status, error_message, created_at)
         VALUES (:id, :machine_id, :payment_hash, :bitspire_event_id,
                 :bitspire_txid, :wire_sats, :fiat_amount, :fiat_code,
                 :exchange_rate, :principal_sats, :fee_sats,
-                :platform_fee_sats, :operator_fee_sats,
+                :platform_fee_sats, :operator_fee_sats, :fee_mismatch_sats,
                 :tx_type, :bills_json, :cassettes_json, :status,
                 :error_message, :created_at)
         """,
@@ -619,6 +625,7 @@ async def create_settlement_idempotent(
             "fee_sats": data.fee_sats,
             "platform_fee_sats": data.platform_fee_sats,
             "operator_fee_sats": data.operator_fee_sats,
+            "fee_mismatch_sats": data.fee_mismatch_sats,
             "tx_type": data.tx_type,
             "bills_json": data.bills_json,
             "cassettes_json": data.cassettes_json,

docs/security-pathway-v1.md

@@ -256,7 +256,7 @@ What we **do not** adopt and why (from the NIP survey):
                        │  Lightning settles
                        ▼
 ┌─────────────────────────────────────────────────────────────────────────────────────────────┐
-│  Settlement attestation (NIP-57-style receipt — kind:9735 or our own kind:21001)            │
+│  Settlement attestation (NIP-57-style receipt — see kind-rotation note in §6 / S3 row)      │
 │                                                                                             │
 │   LNbits publishes (signed by the LNbits server key):                                       │
 │     { kind: 9735,                                                                           │
@@ -303,7 +303,7 @@ None of those need to change. The new layers slot in *above* them.
 | **S0 — Seed‑URL pairing + ATM keypair separation** | Provisioning script generates a fresh `nsec` for the ATM (already does — we just stop overwriting it with the operator's). Operator pastes a one‑shot QR/seed URL containing `{atm_npub, operator_npub, relay_list, signed_delegation_token}` at ATM first boot. | G3 (most of it), G9 | 1 week | None — purely on our side. Use existing NIP‑26 spec. |
 | **S1 — NIP‑40 expiration on all kind‑21000** | Every RPC carries `["expiration", now+5min]`. Handler refuses past‑expiration. ATM clock check on boot (warn if drift > 60s). | G4 | 1–2 days | Relay must support NIP‑40 (most do). |
 | **S2 — NIP‑26 delegation enforcement in nostr‑transport** | Handler parses `delegation` tag, validates sig over conditions, checks conditions match the event, looks up operator pubkey in roster. Reject events without a valid delegation. | G3 (rest), G7 (partially) | 1–2 weeks | LNbits PR upstream (or vendored fork on `aiolabs/lnbits` branch `nostr-transport-nip26`). |
-| **S3 — NIP‑57‑style settlement receipts** | After LNbits internal payment legs complete, publish a signed receipt event per settlement (and per leg if we want leg‑level audit). ATM subscribes; operator dashboard renders receipts side‑by‑side with `dca_settlements`. | G2, G7 | 1–2 weeks | Decide kind: `9735` (semantic abuse for non‑zap) vs. our own kind in `21001`/`21002` range. |
+| **S3 — NIP‑57‑style settlement receipts** | After LNbits internal payment legs complete, publish a signed receipt event per settlement (and per leg if we want leg‑level audit). ATM subscribes; operator dashboard renders receipts side‑by‑side with `dca_settlements`. | G2, G7 | 1–2 weeks | **Kind allocation — DO NOT USE `kind:21001`.** That kind is claimed by CLINK (Offers) — collision caught during the 2026‑06‑02 CLINK primer review. Rotation off 21001 is tracked at `aiolabs/satmachineadmin#44`; target is the aiolabs reserved band **`22000–22099`** per the workspace rule in `~/dev/CLAUDE.md` (§ "Nostr kind allocations — avoid the CLINK band"). The earlier 21001 lock across `aiolabs/lnbits#22`, `aiolabs/satmachineadmin#17`, and the satmachine ATM is **SUPERSEDED** — pick the new kind before any of those land. Reusing `kind:9735` (zap receipt) is also off the table: NIP‑57 semantics don't apply to bitSpire cash‑out settlements. |
 | **S4 — NIP‑78 per‑machine config + fleet roster** | Operator publishes `kind:30078` config + `kind:30000` fleet list. Handler cross‑checks ATM npub ∈ fleet; reads max‑withdraw/fee policy from config. | G1, G9 | 1 week | Define config schema; backwards‑compat path for pre‑NIP‑78 machines. |
 | **S5 — `sender_pubkey` persistence + signed metadata in Payment.extra** | When the dispatcher writes a Payment row, it stamps `Payment.extra.sender_pubkey`, `delegation_root`, and an HMAC over the key fields keyed by the LNbits server's own secret. Mutation post‑write breaks the HMAC. | G2 (DB‑side), G5, G6 | 3–5 days | LNbits PR — fairly localised. |
 | **S6 — Rate limiting + roster‑gated auto‑account** | Auto‑account‑from‑npub only fires if the npub appears in some operator's NIP‑78 fleet OR if an explicit "open enrollment" flag is set. Relay/handler‑level rate limit per pubkey. | G8, G9 | 1 week | LNbits PR. |

fee_transport.py

@@ -0,0 +1,151 @@
+"""
+Fee-config Nostr transport — operator → ATM kind-30078 publish.
+
+Layer 2 of the operator-configurable fee architecture
+(aiolabs/satmachineadmin#37 parent, #39 this layer). Pairs with the
+bitspire consumer at `aiolabs/lamassu-next#57`.
+
+Wire format locked at coord-log §2026-06-01T14:25Z:
+
+  kind = 30078 (NIP-78, replaceable)
+  tags = [
+    ["d", "bitspire-fees:<atm_pubkey_hex>"],
+    ["p", "<atm_pubkey_hex>"],
+  ]
+  content = NIP-44 v2 encrypted JSON of FeeConfigPayload.to_wire_dict()
+  pubkey  = operator pubkey
+  sig     = operator signature
+
+Producer-side invariants (enforced via FeeConfigPayload validators):
+  - cash_*_fee_fraction ≤ 0.15 (cap, mirrored on bitspire consumer)
+  - |total - components sum| < 1e-6 (consistency assert)
+  - schema_version integer ≥ 1
+
+Soft-fail discipline (matches `cassette_transport.publish_to_atm`):
+relay/signer/bunker hiccups log + return None rather than raising,
+so a fee-config trigger from a CRUD endpoint can't break the
+underlying machine create/update on a transient transport failure.
+Hard-raises on configuration errors (cap exceeded, operator has no
+pubkey) since those indicate a bug, not a transient.
+"""
+
+from __future__ import annotations
+
+from loguru import logger
+
+from .models import FeeConfigPayload, FeePayloadComponents, Machine, SuperConfig
+from .nostr_publish import (
+    NostrPublishError,
+    OperatorIdentityMissing,
+    RelayUnavailable,
+    SignerUnavailable,
+    publish_encrypted_kind_30078,
+)
+
+_D_TAG_FEES_PREFIX = "bitspire-fees:"
+
+
+def _atm_hex_pubkey(machine: Machine) -> str:
+    """Canonicalise machine.machine_npub to lowercase hex. Used for both
+    the d-tag suffix and the NIP-44 v2 recipient pubkey. Same shape as
+    cassette_transport's local helper — kept module-local since it's a
+    one-liner over `normalize_public_key` and inlining would invert the
+    abstraction direction (transport-module-knows-about-Machine is
+    correct; nostr_publish doesn't know about Machine)."""
+    from lnbits.utils.nostr import normalize_public_key
+
+    return normalize_public_key(machine.machine_npub).lower()
+
+
+def _fees_d_tag(atm_pubkey_hex: str) -> str:
+    return f"{_D_TAG_FEES_PREFIX}{atm_pubkey_hex}"
+
+
+def build_fee_payload(
+    super_config: SuperConfig, machine: Machine
+) -> FeeConfigPayload:
+    """Compose a FeeConfigPayload from current super-config + per-machine
+    fractions. FeeConfigPayload's validators enforce the cap +
+    consistency invariants — this function constructs and validates
+    in one step; a returned payload is wire-shippable.
+
+    Raises ValueError (via Pydantic) if any directional total exceeds
+    the 0.15 cap. That's a hard error because the upstream API layer
+    (views_api._assert_machine_fee_cap_safe + _assert_super_config_cap_safe)
+    should have rejected the create/update that produced this state.
+    If we reach here with a cap-violating state, something bypassed the
+    API guards and we'd rather refuse-to-publish than ship a malformed
+    event.
+    """
+    super_in = float(super_config.super_cash_in_fee_fraction)
+    super_out = float(super_config.super_cash_out_fee_fraction)
+    op_in = float(machine.operator_cash_in_fee_fraction)
+    op_out = float(machine.operator_cash_out_fee_fraction)
+
+    return FeeConfigPayload(
+        schema_version=1,
+        cash_in_fee_fraction=round(super_in + op_in, 4),
+        cash_out_fee_fraction=round(super_out + op_out, 4),
+        components=FeePayloadComponents(
+            super_cash_in=super_in,
+            super_cash_out=super_out,
+            operator_cash_in=op_in,
+            operator_cash_out=op_out,
+        ),
+    )
+
+
+async def publish_fee_config(
+    machine: Machine,
+    super_config: SuperConfig,
+    operator_user_id: str,
+) -> dict | None:
+    """Build, validate, encrypt, sign, publish the fee-config event for
+    `machine` to the ATM at `machine.machine_npub`.
+
+    Returns the signed event dict on success (caller may log event.id
+    for audit). Returns None on soft-fail (transport-layer errors:
+    relay unreachable, signer offline, bunker timeout) — these are
+    transient and the caller's underlying CRUD operation should succeed
+    independent of publish success. Logs WARNING on soft-fail.
+
+    Raises on hard configuration error:
+      - OperatorIdentityMissing — operator has no Nostr pubkey on file
+        (caller's API layer should refuse the operation before we get
+        here, but we propagate as HTTP 400 if it slips through)
+      - ValueError (from FeeConfigPayload validators) — cap violation
+        or sum/components mismatch, indicates an API-guard bypass
+    """
+    payload = build_fee_payload(super_config, machine)
+    atm_pubkey_hex = _atm_hex_pubkey(machine)
+    try:
+        signed = await publish_encrypted_kind_30078(
+            operator_user_id=operator_user_id,
+            recipient_pubkey_hex=atm_pubkey_hex,
+            d_tag=_fees_d_tag(atm_pubkey_hex),
+            payload=payload.to_wire_dict(),
+            log_context=(
+                f"fee config (machine={machine.id}, "
+                f"cash_in={payload.cash_in_fee_fraction}, "
+                f"cash_out={payload.cash_out_fee_fraction})"
+            ),
+        )
+    except (SignerUnavailable, RelayUnavailable) as exc:
+        logger.warning(
+            f"satmachineadmin: fee-config publish soft-fail for machine "
+            f"{machine.id} ({machine.name or machine.machine_npub[:12]}): "
+            f"{type(exc).__name__}: {exc}. Underlying CRUD operation "
+            "succeeded; operator can re-trigger publish via the next "
+            "machine edit or super-config save."
+        )
+        return None
+    except NostrPublishError as exc:
+        # Truly unexpected transport error — log + soft-fail. We still
+        # don't break the caller's CRUD path; a future publish attempt
+        # (next machine edit / next super edit) will retry.
+        logger.warning(
+            f"satmachineadmin: fee-config publish unexpected transport "
+            f"error for machine {machine.id}: {type(exc).__name__}: {exc}"
+        )
+        return None
+    return signed

migrations.py

@@ -682,9 +682,11 @@ async def m009_split_fee_fractions_by_direction(db):
                             lnbits advisory; option A locked).
 
     Idempotency via column-probe pattern (same shape as m006's rename
-    sweep). The existing `super_config.super_fee_fraction` column is
-    NOT dropped here — deprecated, removed in a follow-up release after
-    callers migrate to the directional fields.
+    sweep). The deprecated `super_config.super_fee_fraction` singleton
+    is backfilled into the new directional fields, then dropped in the
+    same migration — strict-from-the-start per workspace CLAUDE.md
+    "Backwards-compatibility on pre-public-launch code" (v2-bitspire
+    hasn't shipped to public users).
     """
     additions = [
         ("super_config", "super_cash_in_fee_fraction", "DECIMAL(10,4) NOT NULL DEFAULT 0.0000"),
@@ -704,10 +706,22 @@ async def m009_split_fee_fractions_by_direction(db):
             f"ALTER TABLE satoshimachine.{table} ADD COLUMN {col} {coltype}"
         )
 
-    # Backfill super-config directional fractions from the legacy singleton
-    # so the live deployment's super_fee_fraction setting carries forward.
-    # Guarded WHERE clause: only fire when both new fields are still at
-    # their DEFAULT 0 (i.e., this is a first migrate-up, not a repeat).
+    # Backfill + drop the legacy singleton, gated on the column still
+    # existing. Once dropped, a re-run of this migration skips both
+    # steps cleanly.
+    try:
+        await db.fetchone(
+            "SELECT super_fee_fraction FROM satoshimachine.super_config LIMIT 1"
+        )
+        legacy_present = True
+    except Exception:
+        legacy_present = False
+
+    if legacy_present:
+        # Carry the live deployment's super_fee_fraction setting forward
+        # into both directional fields, but only when the operator hasn't
+        # already explicitly set per-direction values (i.e., both are
+        # still at DEFAULT 0).
         await db.execute(
             """
             UPDATE satoshimachine.super_config
@@ -718,3 +732,6 @@ async def m009_split_fee_fractions_by_direction(db):
                AND super_fee_fraction > 0
             """
         )
+        await db.execute(
+            "ALTER TABLE satoshimachine.super_config DROP COLUMN super_fee_fraction"
+        )

models.py

@@ -449,10 +449,6 @@ class TelemetrySnapshot(BaseModel):
 
 class SuperConfig(BaseModel):
     id: str
-    # Deprecated singleton fee fraction — retained for one release while
-    # callers migrate to the per-direction fields below. The new math
-    # (bitspire.py:parse_settlement) only reads the directional fields.
-    super_fee_fraction: float
     super_cash_in_fee_fraction: float = 0.0
     super_cash_out_fee_fraction: float = 0.0
     super_fee_wallet_id: str | None
@@ -460,15 +456,11 @@ class SuperConfig(BaseModel):
 
 
 class UpdateSuperConfigData(BaseModel):
-    # Deprecated; setting either directional field is the supported path.
-    # Writes here continue to apply for one release for migration safety.
-    super_fee_fraction: float | None = None
     super_cash_in_fee_fraction: float | None = None
     super_cash_out_fee_fraction: float | None = None
     super_fee_wallet_id: str | None = None
 
     @validator(
-        "super_fee_fraction",
         "super_cash_in_fee_fraction",
         "super_cash_out_fee_fraction",
     )
@@ -729,3 +721,107 @@ class PublishCassettesPayload(BaseModel):
                 for pos, row in self.positions.items()
             }
         }
+
+
+# =============================================================================
+# Fee-config Nostr payload — operator → ATM (aiolabs/satmachineadmin#39)
+# =============================================================================
+# Locked wire format per coord-log §2026-06-01T14:25Z:
+#   {
+#     "schema_version": 1,
+#     "cash_in_fee_fraction":  super_cash_in  + operator_cash_in,
+#     "cash_out_fee_fraction": super_cash_out + operator_cash_out,
+#     "components": {
+#       "super_cash_in":    float,
+#       "super_cash_out":   float,
+#       "operator_cash_in": float,
+#       "operator_cash_out": float
+#     }
+#   }
+#
+# Producer invariants (refuse-to-publish if violated):
+#   - cash_*_fee_fraction ≤ 0.15 (cap, defense in depth — bitspire
+#     consumer enforces the same)
+#   - |cash_in_fee_fraction - (super_cash_in + operator_cash_in)| < 1e-6
+#   - |cash_out_fee_fraction - (super_cash_out + operator_cash_out)| < 1e-6
+#   - All six fractions in [0.0, 0.15]
+#   - schema_version is integer ≥ 1
+# v1 consumers ignore unknown top-level keys per the locked spec.
+
+
+class FeePayloadComponents(BaseModel):
+    """The producer-mandatory `components` sub-object that splits the
+    summed `cash_*_fee_fraction` totals back into their super + operator
+    halves. Audit + future-promo substrate; consumer-optional in v1."""
+
+    super_cash_in: float
+    super_cash_out: float
+    operator_cash_in: float
+    operator_cash_out: float
+
+
+class FeeConfigPayload(BaseModel):
+    """The decrypted JSON content of a kind-30078 fee-config event
+    (operator → ATM, d-tag `bitspire-fees:<atm_pubkey_hex>`).
+
+    Built from a Machine row + the SuperConfig singleton via
+    `fee_transport.build_fee_payload`. Validates the cap +
+    sum-vs-components consistency at construction time so any caller
+    that holds a FeeConfigPayload instance has a wire-shippable payload.
+    """
+
+    schema_version: int = 1
+    cash_in_fee_fraction: float
+    cash_out_fee_fraction: float
+    components: FeePayloadComponents
+
+    @validator("schema_version")
+    def _schema_version_at_least_v1(cls, v):
+        if v < 1:
+            raise ValueError(f"schema_version must be >= 1, got {v}")
+        return v
+
+    @validator("cash_in_fee_fraction", "cash_out_fee_fraction")
+    def _total_in_unit_range(cls, v):
+        # Imported here rather than at module top to avoid a circular
+        # import (calculations imports nothing from models, but keep the
+        # dependency direction explicit at the call site).
+        from .calculations import MAX_FEE_FRACTION_PER_DIRECTION
+
+        if v < 0 or v > MAX_FEE_FRACTION_PER_DIRECTION:
+            raise ValueError(
+                f"fee fraction must be in [0, {MAX_FEE_FRACTION_PER_DIRECTION}], "
+                f"got {v}"
+            )
+        return round(float(v), 4)
+
+    @validator("components", always=True)
+    def _components_sum_matches_totals(cls, v, values):
+        sum_in = round(v.super_cash_in + v.operator_cash_in, 4)
+        sum_out = round(v.super_cash_out + v.operator_cash_out, 4)
+        total_in = values.get("cash_in_fee_fraction")
+        total_out = values.get("cash_out_fee_fraction")
+        if total_in is not None and abs(total_in - sum_in) > 1e-6:
+            raise ValueError(
+                f"cash_in_fee_fraction={total_in} doesn't match components "
+                f"sum super({v.super_cash_in}) + operator({v.operator_cash_in}) = {sum_in}"
+            )
+        if total_out is not None and abs(total_out - sum_out) > 1e-6:
+            raise ValueError(
+                f"cash_out_fee_fraction={total_out} doesn't match components "
+                f"sum super({v.super_cash_out}) + operator({v.operator_cash_out}) = {sum_out}"
+            )
+        return v
+
+    def to_wire_dict(self) -> dict:
+        return {
+            "schema_version": self.schema_version,
+            "cash_in_fee_fraction": self.cash_in_fee_fraction,
+            "cash_out_fee_fraction": self.cash_out_fee_fraction,
+            "components": {
+                "super_cash_in": self.components.super_cash_in,
+                "super_cash_out": self.components.super_cash_out,
+                "operator_cash_in": self.components.operator_cash_in,
+                "operator_cash_out": self.components.operator_cash_out,
+            },
+        }

nostr_publish.py

@@ -0,0 +1,295 @@
+"""
+Shared kind-30078 (NIP-78 addressable app data) Nostr publish primitives.
+
+Extracted from cassette_transport.py once the second consumer landed
+(fee_transport.py for aiolabs/satmachineadmin#39). Both modules
+share the operator-signer resolution, NIP-44 v2 encrypt/decrypt path,
+event signing, and nostrclient-relay publish path; only the d-tag
+prefix + payload model differ per document type.
+
+Architecture:
+
+  fee_transport / cassette_transport
+       │
+       ▼
+  publish_encrypted_kind_30078  ← high-level wrapper (build event + sign + publish)
+       │
+       ├── resolve_operator_signer
+       ├── nip44_encrypt_via_signer
+       ├── sign_as_operator
+       └── publish_signed_event
+
+`resolve_operator_signer` and the NIP-44 helpers honor the
+transitional LocalSigner → RemoteBunkerSigner cascade (lnbits#17/#18):
+the bunker is the endgame for every operator account on this instance,
+but pre-migration LocalSigner accounts still work via direct-prvkey
+NIP-44 v2 from our hand-rolled `nip44` module.
+
+This module is intentionally domain-agnostic — it knows nothing about
+cassettes, fees, or any specific d-tag prefix. The caller supplies
+the recipient pubkey, the d-tag, and the payload dict.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from typing import Any
+
+from lnbits.core.crud.users import get_account
+from lnbits.core.services.nip46_bunker_client import (
+    NsecBunkerRpcError,
+    NsecBunkerTimeoutError,
+)
+from lnbits.core.signers import resolve_signer
+from lnbits.core.signers.base import (
+    NostrSigner,
+    SignerError,
+    SignerUnavailableError,
+)
+from loguru import logger
+
+from .nip44 import decrypt_from as _nip44_local_decrypt
+from .nip44 import encrypt_for as _nip44_local_encrypt
+
+KIND_NIP78 = 30078
+
+
+# =============================================================================
+# Errors — typed so API endpoints can map to specific HTTP statuses
+# =============================================================================
+
+
+class NostrPublishError(Exception):
+    """Base class for kind-30078 publish errors. Sub-modules
+    (cassette_transport, fee_transport) typically subclass further
+    for domain-specific 'this couldn't be applied' errors that have
+    no analog in the transport layer."""
+
+
+class OperatorIdentityMissing(NostrPublishError):
+    """Operator account has no Nostr pubkey on file, or no signer is
+    available (pre-bunker onboarding — operator hasn't logged in via
+    Nostr-login flow)."""
+
+
+class SignerUnavailable(NostrPublishError):
+    """Resolved signer can't sign server-side (client-side-only signer,
+    or transient bunker unreachability post-lnbits#18). Publish skipped
+    or soft-failed by the caller."""
+
+
+class RelayUnavailable(NostrPublishError):
+    """nostrclient extension isn't installed or its relay manager isn't
+    reachable. Treated as soft-fail by callers; publish skipped + logged."""
+
+
+# =============================================================================
+# Operator signer resolution + NIP-44 v2 encrypt/decrypt
+# =============================================================================
+
+
+async def resolve_operator_signer(operator_user_id: str):
+    """Fetch the operator's account + resolve to a NostrSigner.
+
+    Single source of truth for "give me the signer for this operator,
+    or raise an operator-facing error if we can't." Returns
+    `(account, signer)` so callers that need both (publish path needs
+    `account.pubkey` for the event author and the signer for both
+    encrypt + sign) don't double-fetch.
+
+    Raises:
+      - OperatorIdentityMissing — no account, or no pubkey on file
+      - SignerUnavailable — signer resolve failed, or signer can't sign
+        server-side (ClientSideOnly)
+    """
+    account = await get_account(operator_user_id)
+    if account is None or not account.pubkey:
+        raise OperatorIdentityMissing(
+            f"operator {operator_user_id[:8]}... has no Nostr pubkey on file. "
+            "Onboard via the LNbits Nostr-login flow to publish operator "
+            "config to your ATMs."
+        )
+    try:
+        signer = resolve_signer(account)
+    except SignerError as exc:
+        raise SignerUnavailable(
+            f"signer resolve failed for operator {operator_user_id[:8]}...: {exc}"
+        ) from exc
+    if not signer.can_sign():
+        raise SignerUnavailable(
+            f"operator {operator_user_id[:8]}... has a client-side-only "
+            "signer; server can't sign or NIP-44-encrypt on their behalf. "
+            "Operator must hold their nsec via a NIP-46 bunker (lnbits#18) "
+            "or migrate to a server-signing account."
+        )
+    return account, signer
+
+
+async def sign_as_operator(operator_user_id: str, event: dict) -> dict:
+    """Sign `event` using the operator's signer (LocalSigner or
+    RemoteBunkerSigner). Mutates `event` to add `created_at` (now),
+    `pubkey`, `id`, and `sig`. Returns the signed event.
+
+    Raises typed NostrPublishError subclasses on hard failure (caller
+    maps to HTTP status / decides soft-fail).
+    """
+    _account, signer = await resolve_operator_signer(operator_user_id)
+    # created_at is part of the BIP-340 event-id hash; set before signing.
+    event["created_at"] = int(time.time())
+    try:
+        signed = await signer.sign_event(event)
+    except SignerUnavailableError as exc:
+        raise SignerUnavailable(
+            f"signer unavailable for operator {operator_user_id[:8]}...: {exc}"
+        ) from exc
+    if signed is None:
+        raise NostrPublishError(
+            f"signer returned None for operator {operator_user_id[:8]}... "
+            "— shouldn't be reachable on a server-signing path"
+        )
+    return signed
+
+
+async def nip44_encrypt_via_signer(
+    account, signer: NostrSigner, plaintext: str, peer_pubkey_hex: str
+) -> str:
+    """NIP-44 v2 encrypt via the signer abstraction, with a transitional
+    fallback to direct-prvkey for LocalSigner accounts.
+
+    The bunker (RemoteBunkerSigner) implements `nip44_encrypt` natively —
+    the operator's nsec never leaves the bunker process. LocalSigner's
+    `nip44_encrypt` stub explicitly raises SignerUnavailableError per
+    the post-PR-#38 ABC — the spec is "migrate to bunker." For the
+    transitional window where some operators are still on LocalSigner +
+    their `account.prvkey` is intact, we catch that signal and use our
+    hand-rolled NIP-44 v2 impl against the stored prvkey. Same wire
+    output either way.
+
+    Removed once every operator account on this instance is bunker-
+    backed (S7 fully landed). At that point this helper collapses to
+    `return await signer.nip44_encrypt(plaintext, peer_pubkey_hex)`.
+    """
+    try:
+        return await signer.nip44_encrypt(plaintext, peer_pubkey_hex)
+    except SignerUnavailableError:
+        if account.signer_type == "LocalSigner" and account.prvkey:
+            return _nip44_local_encrypt(plaintext, account.prvkey, peer_pubkey_hex)
+        # ClientSideOnly, or RemoteBunkerSigner with bunker comms failure
+        # at encrypt time — re-raise without wrapping; caller maps it.
+        raise
+
+
+async def nip44_decrypt_via_signer(
+    account, signer: NostrSigner, ciphertext: str, peer_pubkey_hex: str
+) -> str:
+    """Decrypt mirror of `nip44_encrypt_via_signer`. Same LocalSigner
+    transitional fallback."""
+    try:
+        return await signer.nip44_decrypt(ciphertext, peer_pubkey_hex)
+    except SignerUnavailableError:
+        if account.signer_type == "LocalSigner" and account.prvkey:
+            return _nip44_local_decrypt(ciphertext, account.prvkey, peer_pubkey_hex)
+        raise
+
+
+# =============================================================================
+# Relay publish
+# =============================================================================
+
+
+async def publish_signed_event(signed_event: dict) -> None:
+    """Send a signed Nostr event to all relays via the nostrclient
+    extension's singleton RelayManager.
+
+    Lazy import + typed-error so the API can surface "your LNbits doesn't
+    have nostrclient installed" as a 503 rather than a 500. Pattern
+    matches the cross-extension import guards in
+    `lnbits.core.services.users` (nostrmarket / nostrrelay).
+    """
+    try:
+        from nostrclient.router import (  # type: ignore[import-not-found]
+            nostr_client,
+        )
+    except ImportError as exc:
+        raise RelayUnavailable(
+            "nostrclient extension is not installed; kind-30078 publish "
+            "requires it. Install + activate the nostrclient extension on "
+            "this LNbits instance."
+        ) from exc
+    msg = json.dumps(["EVENT", signed_event])
+    nostr_client.relay_manager.publish_message(msg)
+
+
+# =============================================================================
+# High-level: build + encrypt + sign + publish in one call
+# =============================================================================
+
+
+async def publish_encrypted_kind_30078(
+    *,
+    operator_user_id: str,
+    recipient_pubkey_hex: str,
+    d_tag: str,
+    payload: dict[str, Any],
+    log_context: str = "",
+) -> dict:
+    """Build, NIP-44-v2-encrypt, sign-as-operator, and publish a
+    kind-30078 event addressed to `recipient_pubkey_hex` under `d_tag`.
+
+    Centralised so cassette_transport + fee_transport (+ any future
+    operator-pushed document type) share the same wire-format guarantees.
+
+    Returns the signed event dict on success. Raises typed
+    NostrPublishError subclasses on hard failure:
+      - OperatorIdentityMissing → 400: operator hasn't onboarded
+      - SignerUnavailable → 503: signer offline / client-side-only /
+        bunker timeout at encrypt or sign step
+      - RelayUnavailable → 503: nostrclient not installed
+      - NostrPublishError → 500: anything else
+
+    `log_context` is a short string prefixed to the success log line for
+    triage ("cassette", "fee", etc.).
+    """
+    account, signer = await resolve_operator_signer(operator_user_id)
+
+    plaintext = json.dumps(payload, separators=(",", ":"))
+    try:
+        content = await nip44_encrypt_via_signer(
+            account, signer, plaintext, recipient_pubkey_hex
+        )
+    except NsecBunkerTimeoutError as exc:
+        raise SignerUnavailable(
+            f"bunker unreachable while encrypting kind-30078 ({d_tag}) for "
+            f"operator {operator_user_id[:8]}...: {exc}"
+        ) from exc
+    except NsecBunkerRpcError as exc:
+        raise SignerUnavailable(
+            f"bunker rejected nip44_encrypt for operator "
+            f"{operator_user_id[:8]}... (policy / MAC / config issue): {exc}"
+        ) from exc
+    except SignerUnavailableError as exc:
+        raise SignerUnavailable(
+            f"signer cannot nip44-encrypt for operator "
+            f"{operator_user_id[:8]}...: {exc}"
+        ) from exc
+
+    event: dict = {
+        "kind": KIND_NIP78,
+        "tags": [
+            ["d", d_tag],
+            ["p", recipient_pubkey_hex],
+        ],
+        "content": content,
+        # created_at is set inside sign_as_operator before signing.
+    }
+    signed = await sign_as_operator(operator_user_id, event)
+
+    await publish_signed_event(signed)
+    prefix = f"{log_context}: " if log_context else ""
+    logger.info(
+        f"satmachineadmin: {prefix}published kind-30078 to ATM "
+        f"{recipient_pubkey_hex[:12]}... d-tag={d_tag} "
+        f"event_id={signed['id'][:12]}..."
+    )
+    return signed

static/js/index.js

@@ -100,7 +100,11 @@ window.app = Vue.createApp({
       superFeeDialog: {
         show: false,
         saving: false,
-        data: {super_fee_fraction: 0, super_fee_wallet_id: ''}
+        data: {
+          super_cash_in_fee_fraction: 0,
+          super_cash_out_fee_fraction: 0,
+          super_fee_wallet_id: ''
+        }
       },
 
       // UI configuration -----------------------------------------------
@@ -266,6 +270,17 @@ window.app = Vue.createApp({
   },
 
   computed: {
+    superAnyFee() {
+      // Banner styling key — true when either directional super fee is
+      // non-zero, so the banner reads as "active platform fee" instead
+      // of the muted grey "free instance" state.
+      const c = this.superConfig
+      if (!c) return 0
+      return (
+        Number(c.super_cash_in_fee_fraction || 0) +
+        Number(c.super_cash_out_fee_fraction || 0)
+      )
+    },
     walletOptions() {
       // g.user is sometimes null on initial mount in LNbits 1.4 — guard it.
       const wallets = this.g?.user?.wallets || []
@@ -549,7 +564,10 @@ window.app = Vue.createApp({
     // -----------------------------------------------------------------
     openSuperFeeDialog() {
       this.superFeeDialog.data = {
-        super_fee_fraction: this.superConfig?.super_fee_fraction ?? 0,
+        super_cash_in_fee_fraction:
+          this.superConfig?.super_cash_in_fee_fraction ?? 0,
+        super_cash_out_fee_fraction:
+          this.superConfig?.super_cash_out_fee_fraction ?? 0,
         super_fee_wallet_id: this.superConfig?.super_fee_wallet_id || ''
       }
       this.superFeeDialog.show = true
@@ -562,7 +580,8 @@ window.app = Vue.createApp({
         const {data} = await LNbits.api.request(
           'PUT', SUPER_FEE_PATH, null,
           {
-            super_fee_fraction: Number(d.super_fee_fraction),
+            super_cash_in_fee_fraction: Number(d.super_cash_in_fee_fraction),
+            super_cash_out_fee_fraction: Number(d.super_cash_out_fee_fraction),
             super_fee_wallet_id: (d.super_fee_wallet_id || '').trim() || null
           }
         )
@@ -705,7 +724,9 @@ window.app = Vue.createApp({
         location: machine.location || '',
         wallet_id: machine.wallet_id,
         fiat_code: machine.fiat_code,
-        is_active: machine.is_active
+        is_active: machine.is_active,
+        operator_cash_in_fee_fraction: machine.operator_cash_in_fee_fraction ?? 0,
+        operator_cash_out_fee_fraction: machine.operator_cash_out_fee_fraction ?? 0
       }
       this.editMachineDialog.show = true
     },
@@ -723,7 +744,9 @@ window.app = Vue.createApp({
             location: d.location,
             wallet_id: d.wallet_id,
             fiat_code: d.fiat_code,
-            is_active: d.is_active
+            is_active: d.is_active,
+            operator_cash_in_fee_fraction: Number(d.operator_cash_in_fee_fraction) || 0,
+            operator_cash_out_fee_fraction: Number(d.operator_cash_out_fee_fraction) || 0
           }
         )
         const idx = this.machines.findIndex(m => m.id === data.id)
@@ -1475,7 +1498,9 @@ window.app = Vue.createApp({
         wallet_id: null,
         name: '',
         location: '',
-        fiat_code: 'GTQ'
+        fiat_code: 'GTQ',
+        operator_cash_in_fee_fraction: 0,
+        operator_cash_out_fee_fraction: 0
       }
     },
 
@@ -1485,7 +1510,9 @@ window.app = Vue.createApp({
         wallet_id: d.wallet_id,
         name: (d.name || '').trim() || null,
         location: (d.location || '').trim() || null,
-        fiat_code: (d.fiat_code || 'GTQ').trim()
+        fiat_code: (d.fiat_code || 'GTQ').trim(),
+        operator_cash_in_fee_fraction: Number(d.operator_cash_in_fee_fraction) || 0,
+        operator_cash_out_fee_fraction: Number(d.operator_cash_out_fee_fraction) || 0
       }
     },
 

tasks.py

@@ -125,14 +125,14 @@ async def _handle_payment(payment: Payment) -> None:
     # stamp is missing, SettlementInvariantError on any range/sum
     # breach.
     super_config = await get_super_config()
-    super_fee_fraction = float(super_config.super_fee_fraction) if super_config else 0.0
+    assert super_config is not None  # m001 inserts the default singleton
     try:
         data = parse_settlement(
             machine=machine,
             payment_hash=payment.payment_hash,
             wire_sats=payment.sat,
             extra=extra,
-            super_fee_fraction=super_fee_fraction,
+            super_config=super_config,
         )
     except (SettlementMetadataError, SettlementInvariantError) as exc:
         await _record_rejected(payment, machine, exc)
@@ -408,9 +408,9 @@ async def _handle_cassette_state_event(
         CassetteEventDecodeError,
         CassetteEventTransientError,
         CassetteTransportError,
-        _resolve_operator_signer,
         decrypt_and_parse_state_event,
     )
+    from .nostr_publish import resolve_operator_signer
 
     event_raw = event_message.event
     if isinstance(event_raw, str):
@@ -444,7 +444,7 @@ async def _handle_cassette_state_event(
         return
 
     try:
-        account, signer = await _resolve_operator_signer(machine.operator_user_id)
+        account, signer = await resolve_operator_signer(machine.operator_user_id)
     except CassetteTransportError as exc:
         # OperatorIdentityMissing / SignerUnavailable — log + skip.
         logger.warning(

templates/satmachineadmin/index.html

@@ -31,17 +31,19 @@
     <q-banner
       v-if="superConfig"
       class="q-mb-md"
-      :class="superConfig.super_fee_fraction > 0 ? 'bg-blue-1 text-grey-9' : 'bg-grey-2 text-grey-9'">
+      :class="superAnyFee > 0 ? 'bg-blue-1 text-grey-9' : 'bg-grey-2 text-grey-9'">
       <template v-slot:avatar>
-        <q-icon name="account_balance" :color="superConfig.super_fee_fraction > 0 ? 'blue' : 'grey'"></q-icon>
+        <q-icon name="account_balance" :color="superAnyFee > 0 ? 'blue' : 'grey'"></q-icon>
       </template>
       <span :style="{fontWeight: 500}">
         LNbits platform fee:
-        <span :style="{color: '#1976d2'}">${ (superConfig.super_fee_fraction * 100).toFixed(2) }%</span>
-        of each transaction's commission.
+        <span :style="{color: '#1976d2'}">cash-in ${ (superConfig.super_cash_in_fee_fraction * 100).toFixed(2) }%</span>
+        ·
+        <span :style="{color: '#1976d2'}">cash-out ${ (superConfig.super_cash_out_fee_fraction * 100).toFixed(2) }%</span>
+        of each transaction's principal.
       </span>
       <span :style="{opacity: 0.7, fontSize: '0.85em'}" class="q-ml-sm">
-        Your remainder splits per the rules below.
+        Operator's per-machine fee rides on top of these.
       </span>
       <template v-slot:action>
         <q-btn v-if="g?.user?.super_user"
@@ -792,6 +794,22 @@
             hint="Currency the ATM dispenses (GTQ / USD / MXN / etc.)"
             class="q-mb-md"
             dense outlined></q-input>
+
+          <q-input
+            v-model.number="addMachineDialog.data.operator_cash_in_fee_fraction"
+            label="Operator cash-in fee % (decimal, 0..0.15)"
+            hint="Your per-machine cut on cash-in. Sits on top of the platform fee; cap is 15% total per direction."
+            type="number" step="0.0001" min="0" max="0.15"
+            class="q-mb-md"
+            dense outlined></q-input>
+
+          <q-input
+            v-model.number="addMachineDialog.data.operator_cash_out_fee_fraction"
+            label="Operator cash-out fee % (decimal, 0..0.15)"
+            hint="Your per-machine cut on cash-out. Sits on top of the platform fee; cap is 15% total per direction."
+            type="number" step="0.0001" min="0" max="0.15"
+            class="q-mb-md"
+            dense outlined></q-input>
         </q-card-section>
         <q-card-actions align="right" class="text-primary">
           <q-btn flat label="Cancel" v-close-popup></q-btn>
@@ -1226,14 +1244,21 @@
         </q-card-section>
         <q-card-section>
           <p class="text-caption q-mb-md" :style="{opacity: 0.7}">
-            Charged on every operator's commission across the LNbits instance.
-            Operators see this as a read-only banner. Wallet ID is where the
-            collected fee lands; typically a wallet you (the super) own.
+            Charged on every transaction's principal across the LNbits
+            instance. Independent per direction. Each direction's total
+            (super + operator) is capped at 15%. Operators see these as a
+            read-only banner. Wallet ID is where the collected fee lands;
+            typically a wallet you (the super) own.
           </p>
-          <q-input v-model.number="superFeeDialog.data.super_fee_fraction"
-            label="Fee % (decimal, 0..1)"
-            hint="0.30 = 30% of every operator's commission"
-            type="number" step="0.0001" min="0" max="1"
+          <q-input v-model.number="superFeeDialog.data.super_cash_in_fee_fraction"
+            label="Cash-in fee % (decimal, 0..0.15)"
+            hint="0.03 = 3% of principal on cash-in transactions"
+            type="number" step="0.0001" min="0" max="0.15"
+            class="q-mb-md" dense outlined></q-input>
+          <q-input v-model.number="superFeeDialog.data.super_cash_out_fee_fraction"
+            label="Cash-out fee % (decimal, 0..0.15)"
+            hint="0.03 = 3% of principal on cash-out transactions"
+            type="number" step="0.0001" min="0" max="0.15"
             class="q-mb-md" dense outlined></q-input>
           <q-input v-model="superFeeDialog.data.super_fee_wallet_id"
             label="Super fee destination wallet_id"
@@ -1484,6 +1509,16 @@
             dense outlined></q-select>
           <q-input v-model="editMachineDialog.data.fiat_code"
             label="Fiat code" class="q-mb-md" dense outlined></q-input>
+          <q-input v-model.number="editMachineDialog.data.operator_cash_in_fee_fraction"
+            label="Operator cash-in fee % (decimal, 0..0.15)"
+            hint="Sits on top of the platform cash-in fee. Cap 15% total per direction."
+            type="number" step="0.0001" min="0" max="0.15"
+            class="q-mb-md" dense outlined></q-input>
+          <q-input v-model.number="editMachineDialog.data.operator_cash_out_fee_fraction"
+            label="Operator cash-out fee % (decimal, 0..0.15)"
+            hint="Sits on top of the platform cash-out fee. Cap 15% total per direction."
+            type="number" step="0.0001" min="0" max="0.15"
+            class="q-mb-md" dense outlined></q-input>
           <q-toggle v-model="editMachineDialog.data.is_active"
             label="Active (receives settlements)" class="q-mb-md"></q-toggle>
         </q-card-section>

tests/conftest.py

@@ -0,0 +1,32 @@
+"""
+Pytest configuration for the satmachineadmin extension test suite.
+
+Provides a `loguru_capture` fixture for tests that need to verify
+loguru WARN/ERROR side-effects. Loguru attaches its default sink to
+sys.stderr at import time, before pytest's `capsys` wraps stderr, so
+neither `caplog` (stdlib logging only) nor `capsys` reliably sees
+loguru output. The fixture adds a list-sink for the test's duration
+and removes it on teardown.
+"""
+
+from typing import Generator, List
+
+import pytest
+from loguru import logger
+
+
+@pytest.fixture
+def loguru_capture() -> Generator[List[str], None, None]:
+    """Capture loguru log records into a list for the test's duration.
+
+    Usage:
+        def test_warns_on_X(loguru_capture):
+            do_thing_that_warns()
+            assert any("expected message" in msg for msg in loguru_capture)
+    """
+    captured: List[str] = []
+    handler_id = logger.add(
+        captured.append, level="WARNING", format="{level} {message}"
+    )
+    yield captured
+    logger.remove(handler_id)

tests/test_fee_cap_validation.py

@@ -0,0 +1,208 @@
+"""
+Tests for `views_api._assert_machine_fee_cap_safe` and
+`_assert_super_config_cap_safe` (aiolabs/satmachineadmin#38, Layer 1).
+
+Per-direction cap is locked at 15% (super + operator) per coord-log
+§2026-06-01T07:22Z. Both helpers enforce the same cap from the
+opposite direction:
+
+- machine_fee_cap_safe runs at machine create/update; pairs candidate
+  operator fractions against the current super-config
+- super_config_cap_safe runs at super-config update; pairs candidate
+  super fractions against every active machine's operator fractions
+  and names the first offender so the super-admin can fix the
+  triggering machine
+
+Tests monkeypatch the CRUD lookups directly — same shape as
+test_collision_guard.py — so the validators are unit-testable without
+a live LNbits DB.
+"""
+
+import asyncio
+from types import SimpleNamespace
+
+import pytest
+
+from .. import views_api
+from ..views_api import (
+    _assert_machine_fee_cap_safe,
+    _assert_super_config_cap_safe,
+)
+
+
+def _super_config(in_frac: float = 0.0, out_frac: float = 0.0):
+    """Duck-typed super-config row carrying just the two directional fields
+    the cap helpers read."""
+    return SimpleNamespace(
+        super_cash_in_fee_fraction=in_frac,
+        super_cash_out_fee_fraction=out_frac,
+    )
+
+
+def _machine(
+    machine_id: str,
+    op_in: float,
+    op_out: float,
+    name: str | None = None,
+    npub: str = "a" * 64,
+):
+    return SimpleNamespace(
+        id=machine_id,
+        operator_cash_in_fee_fraction=op_in,
+        operator_cash_out_fee_fraction=op_out,
+        name=name,
+        machine_npub=npub,
+    )
+
+
+def _patch_super(monkeypatch, value):
+    async def fake_get():
+        return value
+
+    monkeypatch.setattr(views_api, "get_super_config", fake_get)
+
+
+def _patch_machines(monkeypatch, machines: list):
+    async def fake_list():
+        return machines
+
+    monkeypatch.setattr(views_api, "list_all_active_machines", fake_list)
+
+
+# ---------------------------------------------------------------------------
+# _assert_machine_fee_cap_safe — candidate operator fractions vs current super
+# ---------------------------------------------------------------------------
+
+
+class TestMachineFeeCapSafe:
+    def test_cash_in_cap_exceeded_raises(self, monkeypatch):
+        _patch_super(monkeypatch, _super_config(in_frac=0.10, out_frac=0.05))
+        # 0.10 + 0.06 = 0.16 > 0.15 → reject
+        with pytest.raises(Exception) as exc:
+            asyncio.run(_assert_machine_fee_cap_safe(0.06, 0.05))
+        assert exc.value.status_code == 400
+        assert "cash-in fee cap exceeded" in exc.value.detail
+
+    def test_cash_out_cap_exceeded_raises(self, monkeypatch):
+        _patch_super(monkeypatch, _super_config(in_frac=0.05, out_frac=0.10))
+        # 0.10 + 0.06 = 0.16 > 0.15 → reject
+        with pytest.raises(Exception) as exc:
+            asyncio.run(_assert_machine_fee_cap_safe(0.05, 0.06))
+        assert exc.value.status_code == 400
+        assert "cash-out fee cap exceeded" in exc.value.detail
+
+    def test_at_exact_cap_passes(self, monkeypatch):
+        """The cap check is `>`, not `>=` — operators may set exactly
+        15% on either direction without rejection."""
+        _patch_super(monkeypatch, _super_config(in_frac=0.10, out_frac=0.10))
+        asyncio.run(_assert_machine_fee_cap_safe(0.05, 0.05))
+
+    def test_no_super_config_treats_super_as_zero(self, monkeypatch):
+        """Uninitialised instance (super_config = None) → only operator
+        counts. Cap then degenerates to a pure operator-fee check."""
+        _patch_super(monkeypatch, None)
+        # 0.14 alone is under cap → pass
+        asyncio.run(_assert_machine_fee_cap_safe(0.14, 0.14))
+        # 0.16 alone exceeds cap → reject
+        with pytest.raises(Exception) as exc:
+            asyncio.run(_assert_machine_fee_cap_safe(0.16, 0.05))
+        assert exc.value.status_code == 400
+
+    def test_well_under_cap_passes_silently(self, monkeypatch):
+        _patch_super(monkeypatch, _super_config(in_frac=0.03, out_frac=0.03))
+        # Should not raise.
+        asyncio.run(_assert_machine_fee_cap_safe(0.0333, 0.0777))
+
+    def test_zero_operator_under_zero_super_passes(self, monkeypatch):
+        """Free-charge ATM corner case — operator deliberately sets 0
+        on both directions, super is 0 on both. Cap of 0 ≤ 0.15."""
+        _patch_super(monkeypatch, _super_config(in_frac=0.0, out_frac=0.0))
+        asyncio.run(_assert_machine_fee_cap_safe(0.0, 0.0))
+
+    def test_error_detail_includes_cap_value(self, monkeypatch):
+        _patch_super(monkeypatch, _super_config(in_frac=0.10, out_frac=0.0))
+        with pytest.raises(Exception) as exc:
+            asyncio.run(_assert_machine_fee_cap_safe(0.10, 0.0))
+        # 0.10 + 0.10 = 0.20 > 0.15
+        assert "0.15" in exc.value.detail
+
+
+# ---------------------------------------------------------------------------
+# _assert_super_config_cap_safe — candidate super fractions vs all machines
+# ---------------------------------------------------------------------------
+
+
+class TestSuperConfigCapSafe:
+    def test_offending_machine_raises_and_is_named(self, monkeypatch):
+        """When a super-fee bump pushes one machine over the cap, the
+        rejection names that machine so the super-admin knows which
+        operator's per-machine config blocks the change."""
+        _patch_super(monkeypatch, _super_config(in_frac=0.03, out_frac=0.03))
+        _patch_machines(
+            monkeypatch,
+            [
+                _machine("m1", op_in=0.01, op_out=0.02, name="Cafe A"),
+                _machine("m2", op_in=0.10, op_out=0.02, name="Greedy ATM"),
+            ],
+        )
+        # New super_in = 0.06. m2 has op_in 0.10 → 0.16 > cap.
+        with pytest.raises(Exception) as exc:
+            asyncio.run(_assert_super_config_cap_safe(0.06, None))
+        assert exc.value.status_code == 400
+        assert "Greedy ATM" in exc.value.detail or "m2" in exc.value.detail
+
+    def test_all_machines_under_cap_passes(self, monkeypatch):
+        _patch_super(monkeypatch, _super_config(in_frac=0.03, out_frac=0.03))
+        _patch_machines(
+            monkeypatch,
+            [
+                _machine("m1", op_in=0.05, op_out=0.05, name="Cafe A"),
+                _machine("m2", op_in=0.03, op_out=0.03, name="Cafe B"),
+            ],
+        )
+        # Bump super to 0.08/0.08 → max total = 0.13 + 0.13 = both under cap.
+        asyncio.run(_assert_super_config_cap_safe(0.08, 0.08))
+
+    def test_none_direction_pulls_current_value(self, monkeypatch):
+        """Caller passes new_super_in=None → check uses current super_in
+        value. Confirms partial-update semantics — caller can change
+        cash-out alone without retransmitting cash-in."""
+        _patch_super(monkeypatch, _super_config(in_frac=0.10, out_frac=0.03))
+        _patch_machines(monkeypatch, [_machine("m1", op_in=0.06, op_out=0.0)])
+        # Skipping in (None) but op_in=0.06 + current super_in=0.10 = 0.16 > cap.
+        with pytest.raises(Exception) as exc:
+            asyncio.run(_assert_super_config_cap_safe(None, 0.05))
+        assert exc.value.status_code == 400
+
+    def test_no_machines_passes(self, monkeypatch):
+        """Cap check across an empty fleet is vacuously safe."""
+        _patch_super(monkeypatch, _super_config(in_frac=0.10, out_frac=0.10))
+        _patch_machines(monkeypatch, [])
+        asyncio.run(_assert_super_config_cap_safe(0.12, 0.12))
+
+    def test_no_super_config_with_machines_uses_zero(self, monkeypatch):
+        """Uninitialised super + new fractions → cap check still runs
+        against the candidate new values + each machine's operator
+        fractions."""
+        _patch_super(monkeypatch, None)
+        _patch_machines(
+            monkeypatch,
+            [_machine("m1", op_in=0.10, op_out=0.0, name="Cafe A")],
+        )
+        # 0.06 + 0.10 = 0.16 > cap.
+        with pytest.raises(Exception) as exc:
+            asyncio.run(_assert_super_config_cap_safe(0.06, 0.0))
+        assert exc.value.status_code == 400
+
+    def test_uses_machine_id_when_name_missing(self, monkeypatch):
+        """Machines without a `name` set fall back to the id (or npub
+        prefix) for the error message — operator-actionable in either
+        case."""
+        _patch_super(monkeypatch, _super_config(in_frac=0.03, out_frac=0.03))
+        _patch_machines(
+            monkeypatch,
+            [_machine("unnamed-machine-id", op_in=0.10, op_out=0.0, name=None)],
+        )
+        with pytest.raises(Exception) as exc:
+            asyncio.run(_assert_super_config_cap_safe(0.06, None))
+        assert "unnamed-machine-id" in exc.value.detail

tests/test_fee_mismatch_recording.py

@@ -0,0 +1,179 @@
+"""
+Tests for `dca_settlements.fee_mismatch_sats` Phase-1 observability
+(aiolabs/satmachineadmin#38, coord-log §2026-06-01T07:00Z — option A
+locked: always record, no enforce_fee_match gate).
+
+Each settlement records:
+
+    fee_mismatch_sats = bitspire_fee_sats - (platform_fee_sats + operator_fee_sats)
+
+Positive = bitspire over-reported (claimed more fee than satmachineadmin
+recomputed against principal). Negative = bitspire under-reported.
+Zero = exact match.
+
+Tolerance for the WARN log is `max(1, int(principal_sats * 0.001))` —
+1-sat floor, 0.1% relative ceiling. Sub-tolerance drift records the
+delta silently; over-tolerance drift logs a WARNING. The delta is
+recorded unconditionally regardless of tolerance — sub-tolerance data
+is still useful triage data once aggregated.
+
+Phase 2 (settlement-reject on out-of-tolerance) is a follow-up; this
+layer is observability-only.
+"""
+
+from datetime import datetime
+
+from ..bitspire import parse_settlement
+from ..models import Machine, SuperConfig
+
+
+_NOW = datetime(2026, 6, 1, 12, 0, 0)
+
+
+def _machine(op_out: float = 0.0) -> Machine:
+    return Machine(
+        id="m1",
+        operator_user_id="op1",
+        machine_npub="a" * 64,
+        wallet_id="w1",
+        name="Test",
+        location=None,
+        fiat_code="EUR",
+        is_active=True,
+        operator_cash_in_fee_fraction=0.0,
+        operator_cash_out_fee_fraction=op_out,
+        created_at=_NOW,
+        updated_at=_NOW,
+    )
+
+
+def _super_config(out_frac: float = 0.0) -> SuperConfig:
+    return SuperConfig(
+        id="default",
+        super_cash_in_fee_fraction=0.0,
+        super_cash_out_fee_fraction=out_frac,
+        super_fee_wallet_id="super-wallet",
+        updated_at=_NOW,
+    )
+
+
+def _bitspire_extra(principal_sats: int, fee_sats: int) -> dict:
+    return {
+        "source": "bitspire",
+        "type": "cash_out",
+        "principal_sats": principal_sats,
+        "fee_sats": fee_sats,
+        "fee_fraction": fee_sats / principal_sats if principal_sats else 0.0,
+        "exchange_rate": 0.00001,
+        "fiat_amount": 100.0,
+        "currency": "EUR",
+        "txid": "fake-txid",
+        "nostr_sender_pubkey": "a" * 64,
+    }
+
+
+def _parse(machine, super_cfg, principal_sats, fee_sats):
+    """Helper: build extra + invoke parse_settlement with cash-out wire
+    invariant (wire = principal + fee)."""
+    extra = _bitspire_extra(principal_sats, fee_sats)
+    return parse_settlement(
+        machine=machine,
+        payment_hash="ph_test",
+        wire_sats=principal_sats + fee_sats,
+        extra=extra,
+        super_config=super_cfg,
+    )
+
+
+class TestFeeMismatchSatsRecording:
+    def test_zero_mismatch_when_bitspire_matches_recompute(self):
+        """super=3%, operator=5%, total=8%. Bitspire reports
+        principal=100_000 fee=8_000 → 100_000 * 0.08 = 8_000 → mismatch=0."""
+        machine = _machine(op_out=0.05)
+        super_cfg = _super_config(out_frac=0.03)
+        data = _parse(machine, super_cfg, principal_sats=100_000, fee_sats=8_000)
+        assert data.platform_fee_sats == 3_000
+        assert data.operator_fee_sats == 5_000
+        assert data.fee_mismatch_sats == 0
+
+    def test_positive_mismatch_when_bitspire_over_reports(self):
+        """super=3%, operator=5% → expected=8_000. Bitspire claims 9_000.
+        Delta = +1_000 (over-reported)."""
+        machine = _machine(op_out=0.05)
+        super_cfg = _super_config(out_frac=0.03)
+        data = _parse(machine, super_cfg, principal_sats=100_000, fee_sats=9_000)
+        assert data.fee_mismatch_sats == 1_000
+
+    def test_negative_mismatch_when_bitspire_under_reports(self):
+        """super=3%, operator=5% → expected=8_000. Bitspire claims 7_000.
+        Delta = -1_000 (under-reported)."""
+        machine = _machine(op_out=0.05)
+        super_cfg = _super_config(out_frac=0.03)
+        data = _parse(machine, super_cfg, principal_sats=100_000, fee_sats=7_000)
+        assert data.fee_mismatch_sats == -1_000
+
+    def test_pre_layer3_records_large_delta(self):
+        """Real-world Phase-1 scenario before Layer 3 (lamassu-next#57)
+        ships: ATM hardcodes 7.77% cash-out; operator configures 5%
+        operator + 3% super = 8% total. Bitspire reports
+        100_000 * 0.0777 = 7_770 sats; satmachineadmin recomputes 8_000.
+        Delta is large and visible for triage; behavior unchanged."""
+        machine = _machine(op_out=0.05)
+        super_cfg = _super_config(out_frac=0.03)
+        data = _parse(machine, super_cfg, principal_sats=100_000, fee_sats=7_770)
+        # Expected = 3_000 + 5_000 = 8_000; bitspire claims 7_770.
+        assert data.fee_mismatch_sats == -230
+
+
+class TestFeeMismatchWarningLogging:
+    """Tolerance = max(1, int(principal_sats * 0.001)).
+    For principal=100_000 → tolerance=100. For principal=500 → tolerance=1.
+
+    Uses the `loguru_capture` fixture (defined in conftest.py) to read
+    the WARN log line — pytest's `caplog` only sees stdlib logging,
+    and `capsys` misses loguru's pre-bound stderr sink.
+    """
+
+    def test_within_tolerance_does_not_warn(self, loguru_capture):
+        """1-sat delta at principal=100_000 → tolerance=100 → no warn."""
+        machine = _machine(op_out=0.05)
+        super_cfg = _super_config(out_frac=0.03)
+        data = _parse(machine, super_cfg, principal_sats=100_000, fee_sats=8_001)
+        assert data.fee_mismatch_sats == 1
+        # Still recorded — the delta is small, the WARN is suppressed.
+        assert not any("fee mismatch" in m.lower() for m in loguru_capture)
+
+    def test_outside_tolerance_logs_warning(self, loguru_capture):
+        """101-sat delta at principal=100_000 → tolerance=100 → warns."""
+        machine = _machine(op_out=0.05)
+        super_cfg = _super_config(out_frac=0.03)
+        # bitspire claims 8_101 (= expected 8_000 + 101 over)
+        data = _parse(machine, super_cfg, principal_sats=100_000, fee_sats=8_101)
+        assert data.fee_mismatch_sats == 101
+        assert any("fee mismatch" in m.lower() for m in loguru_capture)
+
+    def test_warning_includes_diagnostic_fields(self, loguru_capture):
+        """WARN log line must carry the fields a triage-time operator
+        needs: bitspire's claim, the expected total, the delta, the
+        principal, both fractions, and tx_type."""
+        machine = _machine(op_out=0.05)
+        super_cfg = _super_config(out_frac=0.03)
+        _parse(machine, super_cfg, principal_sats=100_000, fee_sats=9_000)
+        log_text = "".join(loguru_capture)
+        assert "bitspire_fee_sats=9000" in log_text
+        assert "expected=8000" in log_text
+        assert "delta=1000" in log_text
+        assert "principal=100000" in log_text
+        assert "tx_type=cash_out" in log_text
+
+    def test_one_sat_floor_warns_on_tiny_principal(self, loguru_capture):
+        """At principal=500, tolerance=max(1, 0.5)=1. A 2-sat delta
+        triggers the warning — the floor exists so tiny-principal
+        settlements don't go un-policed."""
+        machine = _machine(op_out=0.05)
+        super_cfg = _super_config(out_frac=0.03)
+        # principal=500 → expected fee = 500 * 0.08 = 40 sats.
+        # Bitspire claims 42 → delta=2. Tolerance=max(1, 0)=1. Warns.
+        data = _parse(machine, super_cfg, principal_sats=500, fee_sats=42)
+        assert data.fee_mismatch_sats == 2
+        assert any("fee mismatch" in m.lower() for m in loguru_capture)

tests/test_fee_publish_triggers.py

@@ -0,0 +1,391 @@
+"""
+Tests for the three views_api trigger points that publish fee config
+to ATMs via fee_transport (aiolabs/satmachineadmin#39 Layer 2):
+
+1. api_create_machine — publish always after create (so ATM unblocks
+   past `awaiting-fees` maintenance, even with default 0/0 operator
+   fees that produce a super-only payload)
+2. api_update_machine — publish only when either operator fee fraction
+   changes (skip on name/location/wallet_id/is_active-only edits)
+3. api_update_super_config — publish to every active machine when
+   either super fraction changes, signed by each machine's operator
+
+Tests monkeypatch `views_api.publish_fee_config` with a recording stub
+to verify the trigger fired (or not) and what arguments it received.
+The publisher itself is exercised by test_fee_transport.py — these
+tests are about the wiring.
+"""
+
+import asyncio
+from datetime import datetime
+
+from .. import views_api
+from ..models import CreateMachineData, Machine, SuperConfig, UpdateMachineData
+
+_NOW = datetime(2026, 6, 1, 12, 0, 0)
+_ATM_PUBKEY_HEX = (
+    "522a4538f1df96508d9ee8b14072344dd4a566acfe03c25a92a39179c6fca891"
+)
+_OP_USER_ID = "ac35c9fc842f40f0a0e9809347cd24d1"
+
+
+def _machine(
+    machine_id: str = "m1",
+    npub: str = _ATM_PUBKEY_HEX,
+    op_in: float = 0.0,
+    op_out: float = 0.0,
+    operator_user_id: str = _OP_USER_ID,
+) -> Machine:
+    return Machine(
+        id=machine_id,
+        operator_user_id=operator_user_id,
+        machine_npub=npub,
+        wallet_id="w1",
+        name=f"machine-{machine_id}",
+        location=None,
+        fiat_code="EUR",
+        is_active=True,
+        operator_cash_in_fee_fraction=op_in,
+        operator_cash_out_fee_fraction=op_out,
+        created_at=_NOW,
+        updated_at=_NOW,
+    )
+
+
+def _super(in_frac: float = 0.03, out_frac: float = 0.03) -> SuperConfig:
+    return SuperConfig(
+        id="default",
+        super_cash_in_fee_fraction=in_frac,
+        super_cash_out_fee_fraction=out_frac,
+        super_fee_wallet_id="super-wallet",
+        updated_at=_NOW,
+    )
+
+
+class _PublishRecorder:
+    """Records every (machine.id, super_in, super_out, operator) tuple
+    publish_fee_config was called with. Drop-in stub for monkeypatching
+    `views_api.publish_fee_config`."""
+
+    def __init__(self):
+        self.calls: list[tuple[str, float, float, float, float, str]] = []
+
+    async def __call__(self, machine, super_config, operator_user_id):
+        self.calls.append(
+            (
+                machine.id,
+                float(super_config.super_cash_in_fee_fraction),
+                float(super_config.super_cash_out_fee_fraction),
+                float(machine.operator_cash_in_fee_fraction),
+                float(machine.operator_cash_out_fee_fraction),
+                operator_user_id,
+            )
+        )
+        return {"id": f"evt_{machine.id}", "kind": 30078}
+
+
+# ---------------------------------------------------------------------------
+# Trigger 1: api_create_machine
+# ---------------------------------------------------------------------------
+
+
+class TestCreateMachineTrigger:
+    def test_publishes_on_create_with_default_operator_fees(self, monkeypatch):
+        """Default 0/0 operator fees — payload carries super-only totals.
+        Publish fires anyway so the ATM gets initial config and can
+        boot past maintenance."""
+        recorder = _PublishRecorder()
+        machine = _machine(op_in=0.0, op_out=0.0)
+
+        async def fake_assert_wallet(*args, **kwargs):
+            return None
+
+        async def fake_assert_collision(*args, **kwargs):
+            return None
+
+        async def fake_assert_fee_cap(*args, **kwargs):
+            return None
+
+        async def fake_create_machine(user_id, data):
+            return machine
+
+        async def fake_get_super():
+            return _super()
+
+        monkeypatch.setattr(views_api, "_assert_wallet_owned_by", fake_assert_wallet)
+        monkeypatch.setattr(views_api, "_assert_no_pubkey_collision", fake_assert_collision)
+        monkeypatch.setattr(views_api, "_assert_machine_fee_cap_safe", fake_assert_fee_cap)
+        monkeypatch.setattr(views_api, "create_machine", fake_create_machine)
+        monkeypatch.setattr(views_api, "get_super_config", fake_get_super)
+        monkeypatch.setattr(views_api, "publish_fee_config", recorder)
+
+        # Build a CreateMachineData + fake User and invoke the endpoint.
+        from types import SimpleNamespace
+
+        data = CreateMachineData(
+            machine_npub=_ATM_PUBKEY_HEX,
+            wallet_id="w1",
+            name="sintra",
+        )
+        user = SimpleNamespace(id=_OP_USER_ID)
+        result = asyncio.run(views_api.api_create_machine(data=data, user=user))
+
+        assert result is machine
+        assert len(recorder.calls) == 1
+        assert recorder.calls[0] == ("m1", 0.03, 0.03, 0.0, 0.0, _OP_USER_ID)
+
+    def test_publishes_on_create_with_nonzero_operator_fees(self, monkeypatch):
+        recorder = _PublishRecorder()
+        machine = _machine(op_in=0.05, op_out=0.05)
+
+        async def passthrough(*args, **kwargs):
+            return None
+
+        async def fake_create_machine(user_id, data):
+            return machine
+
+        async def fake_get_super():
+            return _super(0.03, 0.03)
+
+        monkeypatch.setattr(views_api, "_assert_wallet_owned_by", passthrough)
+        monkeypatch.setattr(views_api, "_assert_no_pubkey_collision", passthrough)
+        monkeypatch.setattr(views_api, "_assert_machine_fee_cap_safe", passthrough)
+        monkeypatch.setattr(views_api, "create_machine", fake_create_machine)
+        monkeypatch.setattr(views_api, "get_super_config", fake_get_super)
+        monkeypatch.setattr(views_api, "publish_fee_config", recorder)
+
+        from types import SimpleNamespace
+
+        data = CreateMachineData(
+            machine_npub=_ATM_PUBKEY_HEX,
+            wallet_id="w1",
+            operator_cash_in_fee_fraction=0.05,
+            operator_cash_out_fee_fraction=0.05,
+        )
+        user = SimpleNamespace(id=_OP_USER_ID)
+        asyncio.run(views_api.api_create_machine(data=data, user=user))
+
+        assert recorder.calls == [("m1", 0.03, 0.03, 0.05, 0.05, _OP_USER_ID)]
+
+    def test_no_super_config_skips_publish(self, monkeypatch):
+        """If the super-config singleton is missing (impossible in
+        practice since m001 inserts it), skip the publish rather than
+        crash the create. Machine still created."""
+        recorder = _PublishRecorder()
+        machine = _machine()
+
+        async def passthrough(*args, **kwargs):
+            return None
+
+        async def fake_create_machine(user_id, data):
+            return machine
+
+        async def fake_get_super():
+            return None
+
+        monkeypatch.setattr(views_api, "_assert_wallet_owned_by", passthrough)
+        monkeypatch.setattr(views_api, "_assert_no_pubkey_collision", passthrough)
+        monkeypatch.setattr(views_api, "_assert_machine_fee_cap_safe", passthrough)
+        monkeypatch.setattr(views_api, "create_machine", fake_create_machine)
+        monkeypatch.setattr(views_api, "get_super_config", fake_get_super)
+        monkeypatch.setattr(views_api, "publish_fee_config", recorder)
+
+        from types import SimpleNamespace
+
+        data = CreateMachineData(machine_npub=_ATM_PUBKEY_HEX, wallet_id="w1")
+        user = SimpleNamespace(id=_OP_USER_ID)
+        result = asyncio.run(views_api.api_create_machine(data=data, user=user))
+
+        assert result is machine
+        assert recorder.calls == []
+
+
+# ---------------------------------------------------------------------------
+# Trigger 2: api_update_machine
+# ---------------------------------------------------------------------------
+
+
+def _wire_update_machine_patches(
+    monkeypatch, existing_machine, updated_machine, recorder
+):
+    """Common setup for api_update_machine tests."""
+
+    async def passthrough(*args, **kwargs):
+        return None
+
+    async def fake_get_machine(machine_id):
+        return existing_machine
+
+    async def fake_update_machine(machine_id, data):
+        return updated_machine
+
+    async def fake_get_super():
+        return _super(0.03, 0.03)
+
+    monkeypatch.setattr(views_api, "_assert_wallet_owned_by", passthrough)
+    monkeypatch.setattr(views_api, "_assert_machine_fee_cap_safe", passthrough)
+    monkeypatch.setattr(views_api, "get_machine", fake_get_machine)
+    monkeypatch.setattr(views_api, "update_machine", fake_update_machine)
+    monkeypatch.setattr(views_api, "get_super_config", fake_get_super)
+    monkeypatch.setattr(views_api, "publish_fee_config", recorder)
+
+
+class TestUpdateMachineTrigger:
+    def test_publishes_when_operator_cash_in_changes(self, monkeypatch):
+        recorder = _PublishRecorder()
+        existing = _machine(op_in=0.05, op_out=0.05)
+        updated = _machine(op_in=0.07, op_out=0.05)
+        _wire_update_machine_patches(monkeypatch, existing, updated, recorder)
+
+        from types import SimpleNamespace
+
+        data = UpdateMachineData(operator_cash_in_fee_fraction=0.07)
+        user = SimpleNamespace(id=_OP_USER_ID)
+        asyncio.run(
+            views_api.api_update_machine(machine_id="m1", data=data, user=user)
+        )
+
+        assert len(recorder.calls) == 1
+        assert recorder.calls[0] == ("m1", 0.03, 0.03, 0.07, 0.05, _OP_USER_ID)
+
+    def test_publishes_when_operator_cash_out_changes(self, monkeypatch):
+        recorder = _PublishRecorder()
+        existing = _machine(op_in=0.05, op_out=0.05)
+        updated = _machine(op_in=0.05, op_out=0.08)
+        _wire_update_machine_patches(monkeypatch, existing, updated, recorder)
+
+        from types import SimpleNamespace
+
+        data = UpdateMachineData(operator_cash_out_fee_fraction=0.08)
+        user = SimpleNamespace(id=_OP_USER_ID)
+        asyncio.run(
+            views_api.api_update_machine(machine_id="m1", data=data, user=user)
+        )
+
+        assert len(recorder.calls) == 1
+        assert recorder.calls[0] == ("m1", 0.03, 0.03, 0.05, 0.08, _OP_USER_ID)
+
+    def test_no_publish_when_only_name_changes(self, monkeypatch):
+        """Name / location / fiat_code / is_active / wallet_id changes
+        don't affect the fee model the ATM enforces — skip the
+        republish to avoid relay churn."""
+        recorder = _PublishRecorder()
+        existing = _machine()
+        updated = _machine()  # same fees
+        _wire_update_machine_patches(monkeypatch, existing, updated, recorder)
+
+        from types import SimpleNamespace
+
+        data = UpdateMachineData(name="new name")
+        user = SimpleNamespace(id=_OP_USER_ID)
+        asyncio.run(
+            views_api.api_update_machine(machine_id="m1", data=data, user=user)
+        )
+
+        assert recorder.calls == []
+
+    def test_no_publish_when_only_is_active_changes(self, monkeypatch):
+        recorder = _PublishRecorder()
+        _wire_update_machine_patches(monkeypatch, _machine(), _machine(), recorder)
+
+        from types import SimpleNamespace
+
+        data = UpdateMachineData(is_active=False)
+        user = SimpleNamespace(id=_OP_USER_ID)
+        asyncio.run(
+            views_api.api_update_machine(machine_id="m1", data=data, user=user)
+        )
+
+        assert recorder.calls == []
+
+
+# ---------------------------------------------------------------------------
+# Trigger 3: api_update_super_config
+# ---------------------------------------------------------------------------
+
+
+class TestSuperConfigUpdateTrigger:
+    def test_publishes_to_every_active_machine_on_super_fraction_change(
+        self, monkeypatch
+    ):
+        """A super-fee change ripples to every active machine since each
+        machine's total = super + machine.operator. Republish per-machine
+        with that machine's operator as the signer (machines owned by
+        different operators sign with different keys)."""
+        recorder = _PublishRecorder()
+        new_super = _super(in_frac=0.04, out_frac=0.04)
+
+        machines = [
+            _machine(machine_id="m1", operator_user_id="op_A"),
+            _machine(machine_id="m2", operator_user_id="op_B", op_in=0.05, op_out=0.07),
+            _machine(machine_id="m3", operator_user_id="op_A", op_in=0.02, op_out=0.02),
+        ]
+
+        async def fake_assert_cap(*args, **kwargs):
+            return None
+
+        async def fake_update_super(data):
+            return new_super
+
+        async def fake_list_active():
+            return machines
+
+        monkeypatch.setattr(
+            views_api, "_assert_super_config_cap_safe", fake_assert_cap
+        )
+        monkeypatch.setattr(views_api, "update_super_config", fake_update_super)
+        monkeypatch.setattr(
+            views_api, "list_all_active_machines", fake_list_active
+        )
+        monkeypatch.setattr(views_api, "publish_fee_config", recorder)
+
+        from types import SimpleNamespace
+        from ..models import UpdateSuperConfigData
+
+        data = UpdateSuperConfigData(super_cash_in_fee_fraction=0.04)
+        user = SimpleNamespace(id="super_admin")
+        asyncio.run(views_api.api_update_super_config(data=data, _user=user))
+
+        assert len(recorder.calls) == 3
+        # Verify each call carries the NEW super fractions + that
+        # machine's operator + own fees
+        assert recorder.calls[0] == ("m1", 0.04, 0.04, 0.0, 0.0, "op_A")
+        assert recorder.calls[1] == ("m2", 0.04, 0.04, 0.05, 0.07, "op_B")
+        assert recorder.calls[2] == ("m3", 0.04, 0.04, 0.02, 0.02, "op_A")
+
+    def test_no_publish_when_only_wallet_id_changes(self, monkeypatch):
+        """Changing super_fee_wallet_id without touching either fraction
+        doesn't affect any ATM's fee model — skip the fleet-wide
+        republish."""
+        recorder = _PublishRecorder()
+        new_super = _super(in_frac=0.03, out_frac=0.03)
+
+        async def fake_assert_cap(*args, **kwargs):
+            return None
+
+        async def fake_update_super(data):
+            return new_super
+
+        async def fake_list_active():
+            raise AssertionError(
+                "list_all_active_machines should not be called when "
+                "no fraction changed"
+            )
+
+        monkeypatch.setattr(
+            views_api, "_assert_super_config_cap_safe", fake_assert_cap
+        )
+        monkeypatch.setattr(views_api, "update_super_config", fake_update_super)
+        monkeypatch.setattr(
+            views_api, "list_all_active_machines", fake_list_active
+        )
+        monkeypatch.setattr(views_api, "publish_fee_config", recorder)
+
+        from types import SimpleNamespace
+        from ..models import UpdateSuperConfigData
+
+        data = UpdateSuperConfigData(super_fee_wallet_id="new-wallet")
+        user = SimpleNamespace(id="super_admin")
+        asyncio.run(views_api.api_update_super_config(data=data, _user=user))
+
+        assert recorder.calls == []

tests/test_fee_transport.py

@@ -0,0 +1,325 @@
+"""
+Tests for `fee_transport.py` and `models.FeeConfigPayload` —
+Layer 2 of the operator-configurable fee architecture
+(aiolabs/satmachineadmin#39).
+
+Three concerns covered:
+
+1. FeeConfigPayload — validators enforce the locked wire-format
+   invariants (cap ≤ 0.15 per direction, components sum matches totals,
+   schema_version ≥ 1).
+2. `build_fee_payload(super_config, machine)` — composes a payload
+   from current DB rows. Wraps construction + validation in one call.
+3. `publish_fee_config(machine, super_config, operator_user_id)` —
+   soft-fail discipline: transport errors log + return None, hard
+   errors (cap-violating state) propagate.
+"""
+
+from datetime import datetime
+
+import pytest
+
+from .. import fee_transport
+from ..fee_transport import build_fee_payload, publish_fee_config
+from ..models import FeeConfigPayload, FeePayloadComponents, Machine, SuperConfig
+from ..nostr_publish import (
+    OperatorIdentityMissing,
+    RelayUnavailable,
+    SignerUnavailable,
+)
+
+_NOW = datetime(2026, 6, 1, 12, 0, 0)
+_ATM_PUBKEY_HEX = (
+    "522a4538f1df96508d9ee8b14072344dd4a566acfe03c25a92a39179c6fca891"
+)
+
+
+def _machine(op_in: float = 0.05, op_out: float = 0.05) -> Machine:
+    return Machine(
+        id="m1",
+        operator_user_id="op1",
+        machine_npub=_ATM_PUBKEY_HEX,
+        wallet_id="w1",
+        name="sintra",
+        location=None,
+        fiat_code="EUR",
+        is_active=True,
+        operator_cash_in_fee_fraction=op_in,
+        operator_cash_out_fee_fraction=op_out,
+        created_at=_NOW,
+        updated_at=_NOW,
+    )
+
+
+def _super(in_frac: float = 0.03, out_frac: float = 0.03) -> SuperConfig:
+    return SuperConfig(
+        id="default",
+        super_cash_in_fee_fraction=in_frac,
+        super_cash_out_fee_fraction=out_frac,
+        super_fee_wallet_id="super-wallet",
+        updated_at=_NOW,
+    )
+
+
+# ---------------------------------------------------------------------------
+# FeeConfigPayload — wire-format validators
+# ---------------------------------------------------------------------------
+
+
+class TestFeeConfigPayloadValidators:
+    def _components(
+        self, s_in: float = 0.03, s_out: float = 0.03, o_in: float = 0.05, o_out: float = 0.05
+    ) -> FeePayloadComponents:
+        return FeePayloadComponents(
+            super_cash_in=s_in,
+            super_cash_out=s_out,
+            operator_cash_in=o_in,
+            operator_cash_out=o_out,
+        )
+
+    def test_well_formed_payload_accepts(self):
+        payload = FeeConfigPayload(
+            schema_version=1,
+            cash_in_fee_fraction=0.08,
+            cash_out_fee_fraction=0.08,
+            components=self._components(),
+        )
+        assert payload.schema_version == 1
+        assert payload.cash_in_fee_fraction == 0.08
+        assert payload.cash_out_fee_fraction == 0.08
+
+    def test_to_wire_dict_round_trips(self):
+        original = FeeConfigPayload(
+            schema_version=1,
+            cash_in_fee_fraction=0.08,
+            cash_out_fee_fraction=0.1077,
+            components=self._components(o_out=0.0777),
+        )
+        wire = original.to_wire_dict()
+        rebuilt = FeeConfigPayload(**wire)
+        assert rebuilt.cash_in_fee_fraction == 0.08
+        assert rebuilt.cash_out_fee_fraction == 0.1077
+        assert rebuilt.components.operator_cash_out == 0.0777
+
+    def test_cap_violation_cash_in_rejects(self):
+        # cap is 0.15 per direction.
+        with pytest.raises(ValueError, match="fee fraction must be in"):
+            FeeConfigPayload(
+                schema_version=1,
+                cash_in_fee_fraction=0.16,
+                cash_out_fee_fraction=0.08,
+                components=self._components(s_in=0.10, o_in=0.06),
+            )
+
+    def test_cap_violation_cash_out_rejects(self):
+        with pytest.raises(ValueError, match="fee fraction must be in"):
+            FeeConfigPayload(
+                schema_version=1,
+                cash_in_fee_fraction=0.08,
+                cash_out_fee_fraction=0.20,
+                components=self._components(s_out=0.10, o_out=0.10),
+            )
+
+    def test_exact_cap_accepted(self):
+        """0.15 exactly is the upper bound — must accept."""
+        FeeConfigPayload(
+            schema_version=1,
+            cash_in_fee_fraction=0.15,
+            cash_out_fee_fraction=0.15,
+            components=self._components(s_in=0.10, s_out=0.10, o_in=0.05, o_out=0.05),
+        )
+
+    def test_inconsistent_total_vs_components_rejects_cash_in(self):
+        """sum(super_cash_in + operator_cash_in) must equal
+        cash_in_fee_fraction within 1e-6."""
+        with pytest.raises(ValueError, match="cash_in_fee_fraction"):
+            FeeConfigPayload(
+                schema_version=1,
+                cash_in_fee_fraction=0.09,  # claims 9%
+                cash_out_fee_fraction=0.08,
+                components=self._components(),  # actually 0.03 + 0.05 = 0.08
+            )
+
+    def test_inconsistent_total_vs_components_rejects_cash_out(self):
+        with pytest.raises(ValueError, match="cash_out_fee_fraction"):
+            FeeConfigPayload(
+                schema_version=1,
+                cash_in_fee_fraction=0.08,
+                cash_out_fee_fraction=0.10,  # claims 10%
+                components=self._components(),  # actually 0.08
+            )
+
+    def test_schema_version_zero_rejects(self):
+        with pytest.raises(ValueError, match="schema_version must be"):
+            FeeConfigPayload(
+                schema_version=0,
+                cash_in_fee_fraction=0.08,
+                cash_out_fee_fraction=0.08,
+                components=self._components(),
+            )
+
+    def test_zero_fractions_accepted(self):
+        """Free-charge ATM — both super + operator at 0 → totals 0."""
+        FeeConfigPayload(
+            schema_version=1,
+            cash_in_fee_fraction=0.0,
+            cash_out_fee_fraction=0.0,
+            components=self._components(s_in=0.0, s_out=0.0, o_in=0.0, o_out=0.0),
+        )
+
+
+# ---------------------------------------------------------------------------
+# build_fee_payload — composition from SuperConfig + Machine
+# ---------------------------------------------------------------------------
+
+
+class TestBuildFeePayload:
+    def test_basic_composition(self):
+        payload = build_fee_payload(_super(0.03, 0.03), _machine(0.05, 0.05))
+        assert payload.cash_in_fee_fraction == 0.08
+        assert payload.cash_out_fee_fraction == 0.08
+        assert payload.components.super_cash_in == 0.03
+        assert payload.components.operator_cash_in == 0.05
+
+    def test_different_directions(self):
+        """Cash-in and cash-out can differ — payload preserves both."""
+        payload = build_fee_payload(_super(0.03, 0.05), _machine(0.0333, 0.0777))
+        assert payload.cash_in_fee_fraction == 0.0633
+        assert payload.cash_out_fee_fraction == 0.1277
+
+    def test_super_only_no_operator(self):
+        """Pre-Layer-2 default — machine has 0/0 operator fees; payload
+        carries super-only totals. This is the 'publish on machine create'
+        path's expected shape."""
+        payload = build_fee_payload(_super(0.03, 0.03), _machine(0.0, 0.0))
+        assert payload.cash_in_fee_fraction == 0.03
+        assert payload.cash_out_fee_fraction == 0.03
+
+    def test_cap_violation_at_build_time_raises(self):
+        """If the API guards were bypassed and the DB has a cap-violating
+        state, build_fee_payload refuses rather than ship a bad payload."""
+        with pytest.raises(ValueError, match="fee fraction must be in"):
+            build_fee_payload(_super(0.10, 0.03), _machine(0.10, 0.0))
+            # 0.10 + 0.10 = 0.20 > 0.15
+
+
+# ---------------------------------------------------------------------------
+# publish_fee_config — soft-fail discipline
+# ---------------------------------------------------------------------------
+
+
+class TestPublishFeeConfigSoftFail:
+    def test_relay_unavailable_returns_none_logs_warning(
+        self, monkeypatch, loguru_capture
+    ):
+        async def fake_publish(**kwargs):
+            raise RelayUnavailable("nostrclient extension is not installed")
+
+        monkeypatch.setattr(
+            fee_transport, "publish_encrypted_kind_30078", fake_publish
+        )
+        import asyncio
+
+        result = asyncio.run(publish_fee_config(_machine(), _super(), "op1"))
+        assert result is None
+        assert any(
+            "soft-fail" in m and "RelayUnavailable" in m for m in loguru_capture
+        )
+
+    def test_signer_unavailable_returns_none_logs_warning(
+        self, monkeypatch, loguru_capture
+    ):
+        async def fake_publish(**kwargs):
+            raise SignerUnavailable("bunker unreachable")
+
+        monkeypatch.setattr(
+            fee_transport, "publish_encrypted_kind_30078", fake_publish
+        )
+        import asyncio
+
+        result = asyncio.run(publish_fee_config(_machine(), _super(), "op1"))
+        assert result is None
+        assert any(
+            "soft-fail" in m and "SignerUnavailable" in m for m in loguru_capture
+        )
+
+    def test_operator_identity_missing_returns_none_logs_warning(
+        self, monkeypatch, loguru_capture
+    ):
+        """OperatorIdentityMissing is a NostrPublishError but not a
+        transport one — currently soft-fails at the same layer. The
+        caller may want to convert this to HTTP 400 in future if the
+        operator-facing UX needs a hard signal, but v1 keeps it soft
+        because a partially-onboarded operator shouldn't crash machine
+        create."""
+
+        async def fake_publish(**kwargs):
+            raise OperatorIdentityMissing("no pubkey on file")
+
+        monkeypatch.setattr(
+            fee_transport, "publish_encrypted_kind_30078", fake_publish
+        )
+        import asyncio
+
+        result = asyncio.run(publish_fee_config(_machine(), _super(), "op1"))
+        assert result is None
+
+    def test_publish_success_returns_signed_event(self, monkeypatch):
+        signed = {
+            "id": "ev1",
+            "kind": 30078,
+            "pubkey": "op_pubkey",
+            "content": "ciphertext",
+            "tags": [["d", f"bitspire-fees:{_ATM_PUBKEY_HEX}"], ["p", _ATM_PUBKEY_HEX]],
+            "created_at": 1780000000,
+            "sig": "ff" * 32,
+        }
+
+        captured = {}
+
+        async def fake_publish(**kwargs):
+            captured.update(kwargs)
+            return signed
+
+        monkeypatch.setattr(
+            fee_transport, "publish_encrypted_kind_30078", fake_publish
+        )
+        import asyncio
+
+        result = asyncio.run(publish_fee_config(_machine(), _super(), "op1"))
+        assert result is signed
+        # Verify d-tag matches the locked spec
+        assert captured["d_tag"] == f"bitspire-fees:{_ATM_PUBKEY_HEX}"
+        assert captured["recipient_pubkey_hex"] == _ATM_PUBKEY_HEX
+        # Payload shape carries components per the §14:25Z lock
+        payload = captured["payload"]
+        assert payload["schema_version"] == 1
+        assert payload["cash_in_fee_fraction"] == 0.08
+        assert "components" in payload
+
+    def test_cap_violation_raises_does_not_soft_fail(self, monkeypatch):
+        """build_fee_payload raises ValueError at construction time on
+        cap-violating state. That's a hard configuration error (API
+        guards bypassed), not a transient transport issue, so it
+        propagates. publish_encrypted_kind_30078 is never reached."""
+
+        called = {"count": 0}
+
+        async def fake_publish(**kwargs):
+            called["count"] += 1
+            return {}
+
+        monkeypatch.setattr(
+            fee_transport, "publish_encrypted_kind_30078", fake_publish
+        )
+        import asyncio
+
+        with pytest.raises(ValueError, match="fee fraction must be in"):
+            asyncio.run(
+                publish_fee_config(
+                    _machine(op_in=0.10, op_out=0.0),
+                    _super(in_frac=0.10, out_frac=0.0),
+                    "op1",
+                )
+            )
+        assert called["count"] == 0

tests/test_operator_split_legs.py

@@ -0,0 +1,150 @@
+"""
+Tests for `allocate_operator_split_legs` (operator's commission-leg
+distribution) and the partial-dispense ratio math in
+`apply_partial_dispense_and_redistribute`.
+
+Both are split-arithmetic concerns that survive the post-#38
+principal-based-math refactor:
+
+  - `allocate_operator_split_legs` slices the operator's share across
+    their commission legs by their per-leg fractions. Function-level,
+    no fee-model coupling.
+  - Partial-dispense ratio math (in distribution.py) preserves the
+    ORIGINAL platform/operator ratio recorded against a settlement at
+    land time when an operator partial-dispenses post-hoc. The ratio
+    comes from the absolute platform_fee_sats / fee_sats recorded on
+    the settlement row, NOT the current super-config fractions — the
+    contract is locked at landing.
+
+Pre-#38 tests for `split_two_stage_commission` lived here; that
+function was removed when the principal-based math landed
+(aiolabs/satmachineadmin#38).
+"""
+
+import pytest
+
+from ..calculations import allocate_operator_split_legs
+
+
+class TestAllocateOperatorSplitLegs:
+    """Operator's remaining share split into commission legs by fraction."""
+
+    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]
+
+    def test_realistic_50_30_20_on_5575(self):
+        amounts = allocate_operator_split_legs(5575, [0.5, 0.3, 0.2])
+        # Plan-scale: 5575 * (0.5, 0.3, 0.2) = (2787.5, 1672.5, 1115)
+        # Last leg absorbs rounding remainders so sum == 5575 exactly.
+        assert sum(amounts) == 5575
+        assert amounts[0] == round(5575 * 0.5)
+        assert amounts[1] == round(5575 * 0.3)
+        # Last leg absorbs the remainder.
+        assert amounts[2] == 5575 - amounts[0] - amounts[1]
+
+    def test_single_leg_full_remainder(self):
+        amounts = allocate_operator_split_legs(7965, [1.0])
+        assert amounts == [7965]
+
+    def test_zero_operator_fee_zeros_all_legs(self):
+        amounts = allocate_operator_split_legs(0, [0.5, 0.3, 0.2])
+        assert amounts == [0, 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 sats split [1/3, 1/3, 1/3] — last leg absorbs the +1 remainder.
+        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,fractions",
+        [
+            (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, fractions):
+        amounts = allocate_operator_split_legs(operator_fee, fractions)
+        assert sum(amounts) == operator_fee
+        assert all(a >= 0 for a in amounts)
+
+
+class TestPartialDispenseSplitRatio:
+    """Partial-dispense recompute (closes #11 H6) must preserve the
+    ORIGINAL platform/operator ratio recorded on the settlement row at
+    land time. Super raising or lowering a global rate post-hoc must
+    NOT retroactively change an existing settlement's share split.
+
+    The math is inlined in `apply_partial_dispense_and_redistribute`
+    (distribution.py) rather than in a standalone function. These tests
+    mirror the inline math so a future refactor doesn't silently change
+    the invariant.
+    """
+
+    def _recompute(self, original_fee, original_platform_fee, new_fee):
+        """Mirror of the ratio math in apply_partial_dispense_and_redistribute."""
+        if original_fee > 0:
+            ratio = original_platform_fee / original_fee
+        else:
+            ratio = 0.0
+        new_platform = round(new_fee * ratio)
+        new_platform = max(0, min(new_platform, new_fee))
+        new_operator = new_fee - new_platform
+        return new_platform, new_operator
+
+    def test_30pct_lands_then_partial(self):
+        # Landed at platform ratio 30/100 = 0.30; new fee = 50.
+        # Original ratio preserved → new_platform = round(50 * 0.30) = 15.
+        new_platform, new_operator = self._recompute(100, 30, 50)
+        assert new_platform == 15
+        assert new_operator == 35
+        assert new_platform + new_operator == 50
+
+    def test_super_changed_rate_doesnt_affect_existing_settlement(self):
+        # Landed with platform=2390, fee=7965 (ratio ≈ 0.30). Super then
+        # bumps the global rate to 50%. Operator partial-dispenses to
+        # 50% gross → new_fee = round(7965 * 0.5) = 3982. The 30% ratio
+        # at land time MUST persist regardless of the new super rate.
+        new_platform, new_operator = self._recompute(7965, 2390, 3982)
+        # Expected with original ratio: round(3982 * 0.30006...) = 1195
+        # With (broken) current rate of 50%: would be 1991 — much higher.
+        assert 1190 <= new_platform <= 1200
+        assert new_platform + new_operator == 3982
+        # Original platform share was ~30%; preserved within rounding.
+        assert abs(new_platform / 3982 - 2390 / 7965) < 0.001
+
+    def test_zero_original_fee_yields_zero_platform(self):
+        new_platform, new_operator = self._recompute(0, 0, 0)
+        assert new_platform == 0
+        assert new_operator == 0
+
+    def test_invariant_sum_equals_new_fee(self):
+        # Random-ish parameter sweep over realistic values.
+        cases = [
+            (100, 30, 50),
+            (100, 0, 50),       # original platform_fee was 0
+            (100, 100, 50),     # original platform_fee was full fee
+            (7965, 2390, 3982),
+            (7965, 7965, 3982),
+            (1_000_000, 333_333, 250_000),
+        ]
+        for orig_comm, orig_plat, new_comm in cases:
+            new_platform, new_operator = self._recompute(
+                orig_comm, orig_plat, new_comm
+            )
+            assert new_platform + new_operator == new_comm, (
+                f"sum invariant violated: {orig_comm=} {orig_plat=} "
+                f"{new_comm=} → {new_platform=} {new_operator=}"
+            )
+            assert 0 <= new_platform <= new_comm

tests/test_principal_based_fees.py

@@ -0,0 +1,270 @@
+"""
+Tests for the post-#38 principal-based fee split:
+
+  - `calculations.split_principal_based(principal_sats, super_frac,
+    operator_frac)` — pure-function math
+  - `bitspire.parse_settlement` — directional dispatch by tx_type
+    ("cash_in" → super_cash_in + operator_cash_in;
+     "cash_out" → super_cash_out + operator_cash_out)
+
+The bug this layer closes: pre-#38 math interpreted super_fee_fraction
+as fraction-of-fee instead of fraction-of-principal, under-paying the
+super by ~13× per cashout. Tests below pin the new math to the
+intended fraction-of-principal model and verify the per-direction
+routing through parse_settlement.
+
+Fee mismatch recording (`fee_mismatch_sats` column, Phase 1
+observability per coord-log §2026-06-01T07:00Z) lands in the next
+commit; those tests live in `test_fee_mismatch_recording.py`.
+"""
+
+from datetime import datetime
+
+import pytest
+
+from ..bitspire import SettlementInvariantError, parse_settlement
+from ..calculations import split_principal_based
+from ..models import Machine, SuperConfig
+
+
+# ---------------------------------------------------------------------------
+# split_principal_based — pure-function math
+# ---------------------------------------------------------------------------
+
+
+class TestSplitPrincipalBased:
+    def test_super_fraction_only(self):
+        """Operator at 0% — super takes exactly super_frac of principal,
+        operator gets 0."""
+        platform, operator = split_principal_based(100_000, 0.03, 0.0)
+        assert platform == 3_000
+        assert operator == 0
+
+    def test_operator_fraction_only(self):
+        """Super at 0% — operator takes exactly operator_frac of
+        principal, platform gets 0."""
+        platform, operator = split_principal_based(100_000, 0.0, 0.05)
+        assert platform == 0
+        assert operator == 5_000
+
+    def test_both_fractions(self):
+        """Both shares independently computed against principal — total
+        is super + operator, not anchored to any fee_sats input."""
+        platform, operator = split_principal_based(100_000, 0.03, 0.05)
+        assert platform == 3_000
+        assert operator == 5_000
+
+    def test_zero_principal_yields_zero_shares(self):
+        platform, operator = split_principal_based(0, 0.03, 0.05)
+        assert platform == 0
+        assert operator == 0
+
+    def test_negative_principal_yields_zero_shares(self):
+        """Defensive: negative principal can't happen in production but
+        the function should not produce negative outputs if it ever does."""
+        platform, operator = split_principal_based(-100, 0.03, 0.05)
+        assert platform == 0
+        assert operator == 0
+
+    def test_rounding_does_not_compound(self):
+        """The two shares round independently — there is no carryover.
+        On a 1_000_000-sat principal with super=0.0333, operator=0.0777,
+        each share rounds against principal individually."""
+        platform, operator = split_principal_based(1_000_000, 0.0333, 0.0777)
+        assert platform == round(1_000_000 * 0.0333)  # 33_300
+        assert operator == round(1_000_000 * 0.0777)  # 77_700
+
+    def test_super_frac_out_of_range_raises(self):
+        with pytest.raises(ValueError, match="super_frac"):
+            split_principal_based(100_000, 1.5, 0.0)
+        with pytest.raises(ValueError, match="super_frac"):
+            split_principal_based(100_000, -0.1, 0.0)
+
+    def test_operator_frac_out_of_range_raises(self):
+        with pytest.raises(ValueError, match="operator_frac"):
+            split_principal_based(100_000, 0.0, 1.5)
+        with pytest.raises(ValueError, match="operator_frac"):
+            split_principal_based(100_000, 0.0, -0.1)
+
+    def test_super_under_payment_bug_regression(self):
+        """Direct regression test for the bug this layer closes.
+
+        Pre-#38 math (deleted): `round(fee_sats * super_fraction)` with
+        fee_sats=8_000 (= 8% of 100_000 principal) and super_fraction=0.03
+        produced platform_fee_sats=240 — ~13× below intent.
+
+        Post-#38 math: split_principal_based(100_000, 0.03, 0.05) gives
+        platform=3_000, which IS the intended 3% of principal."""
+        platform, operator = split_principal_based(100_000, 0.03, 0.05)
+        # Post-#38: super gets intended 3% of principal (3_000 sats)
+        # Pre-#38 would have produced ~240 sats from round(8000 * 0.03).
+        assert platform == 3_000
+
+
+# ---------------------------------------------------------------------------
+# parse_settlement — directional dispatch via tx_type
+# ---------------------------------------------------------------------------
+
+
+def _bitspire_extra(
+    *,
+    tx_type: str = "cash_out",
+    principal_sats: int = 100_000,
+    fee_sats: int = 8_000,
+    exchange_rate: float = 0.00001,
+    fiat_amount: float = 100.0,
+    currency: str = "EUR",
+    nostr_sender_pubkey: str = "a" * 64,
+    extra_overrides: dict | None = None,
+):
+    """Canonical bitspire-stamped Payment.extra dict for tests. Mirrors
+    the shape required by `is_bitspire_payment` + the canonical sat-
+    amount invariants in `_assert_sat_invariants`."""
+    base = {
+        "source": "bitspire",
+        "type": tx_type,
+        "principal_sats": principal_sats,
+        "fee_sats": fee_sats,
+        "fee_fraction": fee_sats / principal_sats if principal_sats else 0.0,
+        "exchange_rate": exchange_rate,
+        "fiat_amount": fiat_amount,
+        "currency": currency,
+        "txid": "fake-txid",
+        "nostr_sender_pubkey": nostr_sender_pubkey,
+    }
+    if extra_overrides:
+        base.update(extra_overrides)
+    return base
+
+
+_NOW = datetime(2026, 6, 1, 12, 0, 0)
+
+
+def _machine(
+    machine_id: str = "m1",
+    machine_npub: str = "a" * 64,
+    op_in: float = 0.0,
+    op_out: float = 0.0,
+    fiat_code: str = "EUR",
+) -> Machine:
+    return Machine(
+        id=machine_id,
+        operator_user_id="op1",
+        machine_npub=machine_npub,
+        wallet_id="w1",
+        name="Test",
+        location=None,
+        fiat_code=fiat_code,
+        is_active=True,
+        operator_cash_in_fee_fraction=op_in,
+        operator_cash_out_fee_fraction=op_out,
+        created_at=_NOW,
+        updated_at=_NOW,
+    )
+
+
+def _super_config(in_frac: float = 0.0, out_frac: float = 0.0) -> SuperConfig:
+    return SuperConfig(
+        id="default",
+        super_cash_in_fee_fraction=in_frac,
+        super_cash_out_fee_fraction=out_frac,
+        super_fee_wallet_id="super-wallet",
+        updated_at=_NOW,
+    )
+
+
+class TestParseSettlementDirectional:
+    def test_cash_out_uses_cash_out_fractions(self):
+        """tx_type='cash_out' must route to super_cash_out +
+        operator_cash_out fractions."""
+        machine = _machine(op_in=0.10, op_out=0.05)
+        super_cfg = _super_config(in_frac=0.10, out_frac=0.03)
+        extra = _bitspire_extra(tx_type="cash_out", principal_sats=100_000)
+
+        data = parse_settlement(
+            machine=machine,
+            payment_hash="ph1",
+            wire_sats=108_000,
+            extra=extra,
+            super_config=super_cfg,
+        )
+        # super_cash_out=0.03, operator_cash_out=0.05 against 100_000 principal
+        assert data.platform_fee_sats == 3_000
+        assert data.operator_fee_sats == 5_000
+        assert data.tx_type == "cash_out"
+
+    def test_cash_in_uses_cash_in_fractions(self):
+        """tx_type='cash_in' must route to super_cash_in +
+        operator_cash_in fractions (not cash_out)."""
+        machine = _machine(op_in=0.04, op_out=0.10)
+        super_cfg = _super_config(in_frac=0.02, out_frac=0.10)
+        extra = _bitspire_extra(tx_type="cash_in", principal_sats=100_000)
+
+        # cash-in wire invariant: wire = principal - fee
+        data = parse_settlement(
+            machine=machine,
+            payment_hash="ph2",
+            wire_sats=92_000,
+            extra=extra,
+            super_config=super_cfg,
+        )
+        # super_cash_in=0.02, operator_cash_in=0.04 against 100_000 principal
+        assert data.platform_fee_sats == 2_000
+        assert data.operator_fee_sats == 4_000
+        assert data.tx_type == "cash_in"
+
+    def test_unknown_tx_type_raises(self):
+        machine = _machine()
+        super_cfg = _super_config()
+        extra = _bitspire_extra(
+            tx_type="cash_out",
+            extra_overrides={"type": "withdrawal"},  # not a known direction
+        )
+        with pytest.raises(SettlementInvariantError, match="unknown tx_type"):
+            parse_settlement(
+                machine=machine,
+                payment_hash="ph3",
+                wire_sats=108_000,
+                extra=extra,
+                super_config=super_cfg,
+            )
+
+    def test_zero_fractions_zero_split(self):
+        """Free-charge ATM: both super + operator at 0 → platform and
+        operator fees are both 0, principal is the full take."""
+        machine = _machine(op_in=0.0, op_out=0.0)
+        super_cfg = _super_config(in_frac=0.0, out_frac=0.0)
+        extra = _bitspire_extra(
+            tx_type="cash_out", principal_sats=100_000, fee_sats=0
+        )
+
+        data = parse_settlement(
+            machine=machine,
+            payment_hash="ph4",
+            wire_sats=100_000,
+            extra=extra,
+            super_config=super_cfg,
+        )
+        assert data.platform_fee_sats == 0
+        assert data.operator_fee_sats == 0
+        assert data.principal_sats == 100_000
+
+    def test_cash_in_does_not_use_cash_out_config(self):
+        """Cross-direction guard: cash-in must NOT pick up cash-out's
+        super or operator fractions even when they're set differently.
+        Pin both directions concretely to prove the dispatch."""
+        machine = _machine(op_in=0.01, op_out=0.10)
+        super_cfg = _super_config(in_frac=0.01, out_frac=0.10)
+        extra = _bitspire_extra(tx_type="cash_in", principal_sats=100_000)
+
+        # cash-in wire invariant: wire = principal - fee
+        data = parse_settlement(
+            machine=machine,
+            payment_hash="ph5",
+            wire_sats=92_000,
+            extra=extra,
+            super_config=super_cfg,
+        )
+        # Cash-in totals = 0.01 + 0.01 = 0.02; not 0.10 + 0.10 = 0.20
+        assert data.platform_fee_sats == 1_000  # 100_000 * 0.01
+        assert data.operator_fee_sats == 1_000  # 100_000 * 0.01

tests/test_two_stage_split.py

@@ -1,214 +0,0 @@
-"""
-Tests for the v2 two-stage commission split (super first, operator remainder).
-
-The plan calls out a verification scenario explicitly:
-  super_fee_fraction=0.30 (i.e. 30%), operator splits [0.5, 0.3, 0.2] on a
-  100-sat fee → super_wallet gets 30, operator legs get 35 / 21 / 14.
-
-Also covers the edge cases: super_fee_fraction=0.0 (no super takes the
-whole fee), super_fee_fraction=1.0 (super takes everything), 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_fraction of the fee; 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_fraction_zero_leaves_all_to_operator(self):
-        platform, operator = split_two_stage_commission(7965, 0.0)
-        assert platform == 0
-        assert operator == 7965
-
-    def test_super_fraction_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("fee_sats", [1, 7, 100, 7965, 1_000_000])
-    @pytest.mark.parametrize("super_fraction", [0.0, 0.1, 0.30, 0.5, 0.777, 1.0])
-    def test_invariant_sum_equals_commission(self, fee_sats, super_fraction):
-        platform, operator = split_two_stage_commission(fee_sats, super_fraction)
-        assert platform + operator == fee_sats
-        assert 0 <= platform <= fee_sats
-        assert 0 <= operator <= fee_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,fractions",
-        [
-            (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, fractions):
-        amounts = allocate_operator_split_legs(operator_fee, fractions)
-        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 fee, super_fee_fraction=0.30, operator splits [0.5, 0.3, 0.2].
-        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_fraction_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_fraction_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
-
-
-class TestPartialDispenseSplitRatio:
-    """The partial-dispense recompute (H6 fix) must preserve the ORIGINAL
-    platform/operator ratio from the landed settlement — NOT re-derive
-    from the current super_fee_fraction.
-
-    These tests cover the math; the actual function lives in distribution.py
-    and is exercised end-to-end via integration testing. Here we verify the
-    invariant a future maintainer should never break.
-    """
-
-    def _recompute(self, original_fee, original_platform_fee, new_fee):
-        """Mirror of the ratio math in apply_partial_dispense_and_redistribute."""
-        if original_fee > 0:
-            ratio = original_platform_fee / original_fee
-        else:
-            ratio = 0.0
-        new_platform = round(new_fee * ratio)
-        new_platform = max(0, min(new_platform, new_fee))
-        new_operator = new_fee - new_platform
-        return new_platform, new_operator
-
-    def test_plan_scenario_30pct_lands_then_partial(self):
-        # Landed at super_fee_fraction=0.30: 100-sat fee → 30 / 70.
-        # Partial-dispense to 50% gross → new_fee = 50.
-        # Original ratio (30/100 = 0.30) preserved.
-        new_platform, new_operator = self._recompute(100, 30, 50)
-        assert new_platform == 15
-        assert new_operator == 35
-        assert new_platform + new_operator == 50
-
-    def test_super_changed_rate_doesnt_affect_existing_settlement(self):
-        # Landed at super_fee_fraction=0.30 (fee 7965, platform 2390).
-        # Super then raises rate to 50% globally. Operator partial-dispenses
-        # to 50% gross → new_fee = 3982 (round(7965 * 0.5)).
-        # Original ratio (2390/7965 ≈ 0.30) MUST still apply, not 50%.
-        new_platform, new_operator = self._recompute(7965, 2390, 3982)
-        # Expected with original ratio: round(3982 * 0.30006...) = 1195
-        # With (broken) current rate of 50%: would be 1991 — much higher.
-        assert 1190 <= new_platform <= 1200
-        assert new_platform + new_operator == 3982
-        # Original platform share was ~30%; preserved within rounding.
-        assert abs(new_platform / 3982 - 2390 / 7965) < 0.001
-
-    def test_zero_original_fee_yields_zero_platform(self):
-        new_platform, new_operator = self._recompute(0, 0, 0)
-        assert new_platform == 0
-        assert new_operator == 0
-
-    def test_invariant_sum_equals_new_fee(self):
-        # Random-ish parameter sweep over realistic values.
-        cases = [
-            (100, 30, 50),
-            (100, 0, 50),       # original platform_fee was 0 (super_fraction=0)
-            (100, 100, 50),     # original platform_fee was 100 (super_fraction=100)
-            (7965, 2390, 3982),
-            (7965, 7965, 3982),
-            (1_000_000, 333_333, 250_000),
-        ]
-        for orig_comm, orig_plat, new_comm in cases:
-            new_platform, new_operator = self._recompute(
-                orig_comm, orig_plat, new_comm
-            )
-            assert new_platform + new_operator == new_comm, (
-                f"sum invariant violated: {orig_comm=} {orig_plat=} "
-                f"{new_comm=} → {new_platform=} {new_operator=}"
-            )
-            assert 0 <= new_platform <= new_comm

views_api.py

@@ -14,6 +14,7 @@ from lnbits.core.models import User
 from lnbits.decorators import check_super_user, check_user_exists
 from lnbits.utils.nostr import normalize_public_key
 
+from .calculations import MAX_FEE_FRACTION_PER_DIRECTION
 from .cassette_transport import (
     CassetteTransportError,
     OperatorIdentityMissing,
@@ -21,6 +22,7 @@ from .cassette_transport import (
     SignerUnavailable,
     publish_to_atm,
 )
+from .fee_transport import publish_fee_config
 from .crud import (
     append_settlement_note,
     count_completed_legs_for_settlement,
@@ -48,6 +50,7 @@ from .crud import (
     get_settlements_for_operator,
     get_stuck_settlements_for_operator,
     get_super_config,
+    list_all_active_machines,
     list_cassette_configs_for_machine,
     lp_is_onboarded,
     replace_commission_splits,
@@ -147,6 +150,105 @@ async def _assert_no_pubkey_collision(machine_npub: str) -> None:
         )
 
 
+async def _assert_machine_fee_cap_safe(
+    operator_in: float,
+    operator_out: float,
+) -> None:
+    """Reject create/update if (super_X + operator_X) > 0.15 for either
+    direction. Locked at 15% per coord-log §2026-06-01T07:22Z; defense in
+    depth — the bitspire consumer enforces the same cap on the wire-format
+    side (aiolabs/lamassu-next#57).
+
+    Fetches the current super-config singleton to pair against the
+    candidate per-machine fractions. NULL super-config (uninitialised
+    instance) treats super contribution as 0 — the cap then degenerates
+    to a pure operator-fee check.
+    """
+    super_config = await get_super_config()
+    super_in = (
+        float(super_config.super_cash_in_fee_fraction) if super_config else 0.0
+    )
+    super_out = (
+        float(super_config.super_cash_out_fee_fraction) if super_config else 0.0
+    )
+    # Fields are stored as DECIMAL(10,4) and Pydantic validators round to
+    # 4 decimals on the way in, so the source-of-truth precision is 1e-4.
+    # Round the float-arithmetic sum to that precision before comparison so
+    # `0.10 + 0.05 = 0.15000000000000002` (IEEE 754) doesn't trip the cap.
+    total_in = round(super_in + operator_in, 4)
+    total_out = round(super_out + operator_out, 4)
+    if total_in > MAX_FEE_FRACTION_PER_DIRECTION:
+        raise HTTPException(
+            HTTPStatus.BAD_REQUEST,
+            (
+                f"cash-in fee cap exceeded: super {super_in:.4f} + operator "
+                f"{operator_in:.4f} = {total_in:.4f} > {MAX_FEE_FRACTION_PER_DIRECTION}"
+            ),
+        )
+    if total_out > MAX_FEE_FRACTION_PER_DIRECTION:
+        raise HTTPException(
+            HTTPStatus.BAD_REQUEST,
+            (
+                f"cash-out fee cap exceeded: super {super_out:.4f} + operator "
+                f"{operator_out:.4f} = {total_out:.4f} > {MAX_FEE_FRACTION_PER_DIRECTION}"
+            ),
+        )
+
+
+async def _assert_super_config_cap_safe(
+    new_super_in: float | None,
+    new_super_out: float | None,
+) -> None:
+    """Reject super-config update if any active machine's
+    (new_super + operator) > 0.15 for either direction. Same cap policy
+    as _assert_machine_fee_cap_safe but checked across the fleet because
+    a super update affects every machine.
+
+    `None` for a direction means "no change" — pulls the current value
+    from super-config so the cap check still runs against the resulting
+    post-update state.
+    """
+    current = await get_super_config()
+    effective_in = (
+        float(new_super_in)
+        if new_super_in is not None
+        else (float(current.super_cash_in_fee_fraction) if current else 0.0)
+    )
+    effective_out = (
+        float(new_super_out)
+        if new_super_out is not None
+        else (float(current.super_cash_out_fee_fraction) if current else 0.0)
+    )
+    machines = await list_all_active_machines()
+    for m in machines:
+        op_in = float(m.operator_cash_in_fee_fraction)
+        op_out = float(m.operator_cash_out_fee_fraction)
+        # Round to DECIMAL(10,4) precision — see _assert_machine_fee_cap_safe
+        # for the IEEE 754 motivation.
+        total_in = round(effective_in + op_in, 4)
+        total_out = round(effective_out + op_out, 4)
+        if total_in > MAX_FEE_FRACTION_PER_DIRECTION:
+            raise HTTPException(
+                HTTPStatus.BAD_REQUEST,
+                (
+                    f"super cash-in fee {effective_in:.4f} would exceed cap "
+                    f"on machine {m.id} ({m.name or m.machine_npub[:12]}): "
+                    f"+ operator {op_in:.4f} = "
+                    f"{total_in:.4f} > {MAX_FEE_FRACTION_PER_DIRECTION}"
+                ),
+            )
+        if total_out > MAX_FEE_FRACTION_PER_DIRECTION:
+            raise HTTPException(
+                HTTPStatus.BAD_REQUEST,
+                (
+                    f"super cash-out fee {effective_out:.4f} would exceed cap "
+                    f"on machine {m.id} ({m.name or m.machine_npub[:12]}): "
+                    f"+ operator {op_out:.4f} = "
+                    f"{total_out:.4f} > {MAX_FEE_FRACTION_PER_DIRECTION}"
+                ),
+            )
+
+
 # =============================================================================
 # Machines
 # =============================================================================
@@ -158,7 +260,17 @@ async def api_create_machine(
 ) -> Machine:
     await _assert_wallet_owned_by(data.wallet_id, user.id)
     await _assert_no_pubkey_collision(data.machine_npub)
+    await _assert_machine_fee_cap_safe(
+        data.operator_cash_in_fee_fraction,
+        data.operator_cash_out_fee_fraction,
+    )
     machine = await create_machine(user.id, data)
+    # Layer 2 (#39): publish initial fee config to the ATM so it can
+    # unblock past its `awaiting-fees` maintenance gate. Soft-fails on
+    # transport errors — machine creation has already succeeded.
+    super_config = await get_super_config()
+    if super_config is not None:
+        await publish_fee_config(machine, super_config, user.id)
     return machine
 
 
@@ -194,9 +306,38 @@ async def api_update_machine(
         raise HTTPException(HTTPStatus.NOT_FOUND, "Machine not found")
     if data.wallet_id is not None:
         await _assert_wallet_owned_by(data.wallet_id, user.id)
+    # Cap check against post-update state — partial PATCH semantics:
+    # unset directional fields keep the machine's current value.
+    if (
+        data.operator_cash_in_fee_fraction is not None
+        or data.operator_cash_out_fee_fraction is not None
+    ):
+        candidate_in = (
+            data.operator_cash_in_fee_fraction
+            if data.operator_cash_in_fee_fraction is not None
+            else float(machine.operator_cash_in_fee_fraction)
+        )
+        candidate_out = (
+            data.operator_cash_out_fee_fraction
+            if data.operator_cash_out_fee_fraction is not None
+            else float(machine.operator_cash_out_fee_fraction)
+        )
+        await _assert_machine_fee_cap_safe(candidate_in, candidate_out)
     updated = await update_machine(machine_id, data)
     if updated is None:
         raise HTTPException(HTTPStatus.NOT_FOUND, "Machine not found")
+    # Layer 2 (#39): if either operator fee fraction changed, publish a
+    # fresh kind-30078 to the ATM so it picks up the new total. Skip
+    # otherwise — name/location/wallet_id/is_active edits don't change
+    # the fee model the ATM enforces.
+    fees_changed = (
+        data.operator_cash_in_fee_fraction is not None
+        or data.operator_cash_out_fee_fraction is not None
+    )
+    if fees_changed:
+        super_config = await get_super_config()
+        if super_config is not None:
+            await publish_fee_config(updated, super_config, user.id)
     return updated
 
 
@@ -808,11 +949,29 @@ async def api_update_super_config(
     commission, plus the destination wallet for collecting it. The fee is
     enforced before the operator's own commission_splits ruleset fires
     (see distribution.process_settlement)."""
+    await _assert_super_config_cap_safe(
+        data.super_cash_in_fee_fraction,
+        data.super_cash_out_fee_fraction,
+    )
     config = await update_super_config(data)
     if config is None:
         raise HTTPException(
             HTTPStatus.INTERNAL_SERVER_ERROR, "Failed to update super config"
         )
+    # Layer 2 (#39): a super-fee change ripples to every active machine
+    # since each machine's total = super + machine.operator. Republish
+    # per-machine with that machine's operator as the signer.
+    # Soft-fails per machine independently; partial success is acceptable
+    # (the operator whose publish failed can re-trigger via a machine
+    # edit). Skip if neither directional fraction was touched in this
+    # update (e.g. caller only changed super_fee_wallet_id).
+    super_fractions_changed = (
+        data.super_cash_in_fee_fraction is not None
+        or data.super_cash_out_fee_fraction is not None
+    )
+    if super_fractions_changed:
+        for machine in await list_all_active_machines():
+            await publish_fee_config(machine, config, machine.operator_user_id)
     return config