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 .nostr_publish import (
-from .nip44 import encrypt_for as _nip44_local_encrypt
+    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):
+class CassetteTransportError(NostrPublishError):
-    """Generic transport-layer error. Subclasses distinguish failure modes
+    """Generic cassette-transport error. Subclasses distinguish failure
-    so the API can surface meaningful HTTP statuses + the consumer task
+    modes so the API can surface meaningful HTTP statuses + the consumer
-    can log + skip without crashing."""
+    task can log + skip without crashing.
 
-
+    Bridges back-compat with pre-extraction callers that catch this
-class OperatorIdentityMissing(CassetteTransportError):
+    class — now equivalent to NostrPublishError plus the two consumer-
-    """Operator account has no Nostr pubkey on file, or no signer is
+    side decode/transient distinctions below.
-    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."""
 
 
 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:
+    audit). Raises NostrPublishError subclasses (re-exported here as
-      - OperatorIdentityMissing → 400: operator hasn't onboarded
+    CassetteTransportError, OperatorIdentityMissing, SignerUnavailable,
-      - SignerUnavailable → 503: signer offline / client-side-only / bunker
+    RelayUnavailable) on hard failures.
-        timeout at the encrypt or sign step
-      - RelayUnavailable → 503: nostrclient not installed
-      - CassetteTransportError → 500: anything else
     """
     atm_pubkey_hex = _atm_hex_pubkey(machine)
-
+    signed = await publish_encrypted_kind_30078(
-    # Single fetch + resolve — same signer is used for both encrypt and sign.
+        operator_user_id=operator_user_id,
-    account, signer = await _resolve_operator_signer(operator_user_id)
+        recipient_pubkey_hex=atm_pubkey_hex,
-
+        d_tag=_config_d_tag(atm_pubkey_hex),
-    # NIP-44 v2 encrypt the wire payload. Bunker round-trip on
+        payload=payload.to_wire_dict(),
-    # RemoteBunkerSigner; direct prvkey on LocalSigner (transitional).
+        log_context=(
-    plaintext = json.dumps(payload.to_wire_dict(), separators=(",", ":"))
+            f"cassette config (machine={machine.id}, "
-    try:
+            f"positions={sorted(payload.positions.keys())})"
-        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())})"
     )
     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:

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

models.py

@@ -721,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

tasks.py

@@ -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(

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

views_api.py

@@ -22,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,
@@ -264,6 +265,12 @@ async def api_create_machine(
         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
 
 
@@ -319,6 +326,18 @@ async def api_update_machine(
     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
 
 
@@ -939,6 +958,20 @@ async def api_update_super_config(
         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