feat(cassettes): schema and models for operator operations

First piece of the v2 wire (bitspire ADR-004). The operator stops
publishing counts and starts publishing what it DID; the machine, which
holds the notes, keeps the running total. A value with one writer cannot
be clobbered, which is the whole point: the absolute-count wire let a
form loaded before a dispense discard that dispense when published, and
nothing in an addressable event can tell the loser it lost.

m013 adds cassette_ops, append-only. The id is minted here and is the
idempotency key the machine dedups on, because a delta applied twice is
wrong and addressable events are re-delivered on reconnect. acked_at is
set when the machine reports that id back, which is the only
acknowledgement this transport can carry.

The models enforce that an op carries exactly the one field its type
means, so an instance is publishable by construction — the same contract
FeeConfigPayload has — and nulls never reach the wire for the machine to
disambiguate. recount is the only absolute, deliberately: it is what an
operator opening a bay and counting actually does, and it stays
auditable as its own act rather than looking like a stale form.

Vocabulary mirrors lamassu-server's cash_unit_operation_type.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

models.py

@@ -7,7 +7,7 @@
 
 from datetime import datetime
 
-from pydantic import BaseModel, validator
+from pydantic import BaseModel, root_validator, validator
 
 # =============================================================================
 # Machines — one row per bitSpire ATM, owned by exactly one operator.
@@ -768,6 +768,170 @@ class PublishCassettesPayload(BaseModel):
         }
 
 
+# =============================================================================
+# Cassette operations — operator → ATM v2 (bitspire ADR-004)
+# =============================================================================
+# The operator no longer publishes counts. It publishes what it DID, and the
+# machine — which holds the notes — keeps the running total. A value with one
+# writer cannot be clobbered, which is the point: the old absolute-count wire
+# let a form loaded before a dispense discard that dispense when published, and
+# nothing in an addressable event can tell the loser it lost.
+#
+# Wire shape (kind-30078 content, NIP-44 v2 encrypted, schema_version 2):
+#   {
+#     "schema_version": 2,
+#     "ops": [
+#       {"id": "<uuid>", "at": 1790106060, "type": "refill",
+#        "position": 2, "bills": 100},
+#       {"id": "<uuid>", "at": 1790106061, "type": "empty", "position": 3},
+#       {"id": "<uuid>", "at": 1790106062, "type": "recount",
+#        "position": 1, "count": 37},
+#       {"id": "<uuid>", "at": 1790106063, "type": "set_denomination",
+#        "position": 1, "denomination": 50}
+#     ]
+#   }
+#
+# `ops` is a WINDOW of recent operations, not just the newest. An event the
+# machine missed is carried again by the next one, so the channel heals itself
+# without the operator noticing. `id` is the idempotency key: deltas are not
+# idempotent and addressable events are re-delivered on reconnect, so the
+# machine records what it applied and ignores repeats.
+#
+# The vocabulary mirrors lamassu-server's cash_unit_operation_type
+# (refill / empty / count-change), which is where the ancestor of this fleet
+# landed after the same problem.
+
+
+CASSETTE_OP_TYPES = ("refill", "empty", "recount", "set_denomination")
+
+
+class CassetteOp(BaseModel):
+    """One operator intent against one bay, as stored and as published.
+
+    Exactly one of bills/count/denomination is meaningful, decided by op_type:
+      - refill           → bills, the number of notes ADDED (a delta)
+      - empty            → none; the bay was emptied
+      - recount          → count, an absolute the operator physically counted
+      - set_denomination → denomination, what is now loaded in that bay
+
+    `recount` is the only absolute, and deliberately so: it is what an operator
+    opening a bay and counting actually does, and it is auditable as a distinct
+    act rather than being indistinguishable from a stale form.
+    """
+
+    id: str
+    machine_id: str
+    position: int
+    op_type: str
+    bills: int | None = None
+    count: int | None = None
+    denomination: int | None = None
+    created_at: datetime
+    created_by: str | None = None
+    acked_at: datetime | None = None
+
+    @validator("op_type")
+    def _known_op_type(cls, v):
+        if v not in CASSETTE_OP_TYPES:
+            raise ValueError(f"op_type must be one of {CASSETTE_OP_TYPES}, got {v!r}")
+        return v
+
+    @validator("position")
+    def _position_positive(cls, v):
+        if v <= 0:
+            raise ValueError(f"position must be > 0, got {v}")
+        return v
+
+    def to_wire_dict(self) -> dict:
+        """The published form. Drops the fields this op_type does not use, so
+        the machine never has to guess which of three nullable columns applies."""
+        out: dict = {
+            "id": self.id,
+            "at": int(self.created_at.timestamp()),
+            "type": self.op_type,
+            "position": self.position,
+        }
+        if self.op_type == "refill":
+            out["bills"] = self.bills
+        elif self.op_type == "recount":
+            out["count"] = self.count
+        elif self.op_type == "set_denomination":
+            out["denomination"] = self.denomination
+        return out
+
+
+class CreateCassetteOpData(BaseModel):
+    """Operator submits one operation from the dashboard.
+
+    Validated per type here rather than at the endpoint so an instance is
+    always publishable, matching FeeConfigPayload's contract.
+    """
+
+    position: int
+    op_type: str
+    bills: int | None = None
+    count: int | None = None
+    denomination: int | None = None
+
+    @validator("op_type")
+    def _known_op_type(cls, v):
+        if v not in CASSETTE_OP_TYPES:
+            raise ValueError(f"op_type must be one of {CASSETTE_OP_TYPES}, got {v!r}")
+        return v
+
+    @validator("position")
+    def _position_positive(cls, v):
+        if v <= 0:
+            raise ValueError(f"position must be > 0, got {v}")
+        return v
+
+    @validator("bills")
+    def _bills_positive(cls, v):
+        if v is not None and v <= 0:
+            raise ValueError("bills must be > 0 (a refill adds notes)")
+        return v
+
+    @validator("count")
+    def _count_non_negative(cls, v):
+        if v is not None and v < 0:
+            raise ValueError("count must be >= 0")
+        return v
+
+    @validator("denomination")
+    def _denomination_positive(cls, v):
+        if v is not None and v <= 0:
+            raise ValueError("denomination must be > 0")
+        return v
+
+    @root_validator(skip_on_failure=True)
+    def _field_matches_type(cls, values):
+        required = {
+            "refill": "bills",
+            "recount": "count",
+            "set_denomination": "denomination",
+            "empty": None,
+        }[values.get("op_type")]
+        if required is not None and values.get(required) is None:
+            raise ValueError(f"{values['op_type']} requires `{required}`")
+        for field in ("bills", "count", "denomination"):
+            if field != required and values.get(field) is not None:
+                raise ValueError(f"{values['op_type']} must not carry `{field}`")
+        return values
+
+
+class PublishCassetteOpsPayload(BaseModel):
+    """The decrypted content of a v2 operator → ATM cassette event."""
+
+    schema_version: int = 2
+    ops: list[CassetteOp]
+
+    def to_wire_dict(self) -> dict:
+        return {
+            "schema_version": self.schema_version,
+            "ops": [op.to_wire_dict() for op in self.ops],
+        }
+
+
 # =============================================================================
 # Fee-config Nostr payload — operator → ATM (aiolabs/satmachineadmin#39)
 # =============================================================================