diff --git a/migrations.py b/migrations.py
index 6ede0eb..52aebf3 100644
--- a/migrations.py
+++ b/migrations.py
@@ -860,3 +860,44 @@ async def m012_add_max_cash_in_sats(db):
     await db.execute(
         "ALTER TABLE spirekeeper.super_config ADD COLUMN max_cash_in_sats INTEGER"
     )
+
+
+async def m013_add_cassette_ops(db):
+    """Cassette operations — the operator→ATM v2 wire (aiolabs/bitspire ADR-004).
+
+    Until now the operator published absolute counts and the ATM applied them
+    outright. Both sides wrote the same value over a transport that never tells
+    a writer it lost, so a dashboard form loaded before a dispense would
+    silently discard that dispense when published. The fix is to stop the
+    operator writing counts at all: it publishes OPERATIONS and the machine,
+    which holds the physical notes, owns the running count.
+
+    Each row here is one operator intent — a refill, an empty, a recount, a
+    denomination change. `id` is minted here and is the idempotency key the ATM
+    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
+    the id back in its state document, which is the only acknowledgement this
+    transport can carry.
+
+    Kept append-only on purpose: the published window is a slice of this table,
+    and an operation the machine has not yet acknowledged must stay publishable.
+    """
+    await db.execute(f"""
+        CREATE TABLE IF NOT EXISTS spirekeeper.cassette_ops (
+            id TEXT PRIMARY KEY,
+            machine_id TEXT NOT NULL,
+            position INTEGER NOT NULL,
+            op_type TEXT NOT NULL,
+            bills INTEGER,
+            count INTEGER,
+            denomination INTEGER,
+            created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
+            created_by TEXT,
+            acked_at TIMESTAMP
+        );
+        """)
+    # The publisher reads the most recent N for a machine on every publish.
+    await db.execute(
+        "CREATE INDEX IF NOT EXISTS cassette_ops_machine_idx "
+        "ON cassette_ops (machine_id, created_at DESC)"
+    )
diff --git a/models.py b/models.py
index 5a4e529..9112340 100644
--- a/models.py
+++ b/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)
 # =============================================================================
diff --git a/tests/test_cassette_ops.py b/tests/test_cassette_ops.py
new file mode 100644
index 0000000..8187fa2
--- /dev/null
+++ b/tests/test_cassette_ops.py
@@ -0,0 +1,142 @@
+"""
+Tests for the v2 cassette-operations models (bitspire ADR-004).
+
+The operator no longer publishes counts; it publishes operations and the
+machine keeps the running total. These cover the pure pieces: per-type field
+validation, and the wire shape the publisher ships.
+
+A CreateCassetteOpData instance is meant to be publishable by construction —
+same contract as FeeConfigPayload — so the type/field agreement is enforced in
+the model rather than at the endpoint.
+"""
+
+from datetime import datetime, timezone
+
+import pytest
+from pydantic import ValidationError
+
+from ..models import (
+    CASSETTE_OP_TYPES,
+    CassetteOp,
+    CreateCassetteOpData,
+    PublishCassetteOpsPayload,
+)
+
+AT = datetime.fromtimestamp(1790106060, timezone.utc)
+
+
+def op(**kw) -> CassetteOp:
+    base = {"id": "op-1", "machine_id": "m1", "position": 2, "created_at": AT}
+    return CassetteOp(**{**base, **kw})
+
+
+class TestCreateCassetteOpData:
+    def test_accepts_one_of_each_type(self):
+        CreateCassetteOpData(position=2, op_type="refill", bills=100)
+        CreateCassetteOpData(position=3, op_type="empty")
+        CreateCassetteOpData(position=1, op_type="recount", count=37)
+        CreateCassetteOpData(position=1, op_type="set_denomination", denomination=50)
+
+    @pytest.mark.parametrize(
+        "kwargs",
+        [
+            {"position": 2, "op_type": "refill"},
+            {"position": 1, "op_type": "recount"},
+            {"position": 1, "op_type": "set_denomination"},
+        ],
+    )
+    def test_rejects_a_type_missing_its_field(self, kwargs):
+        with pytest.raises(ValidationError):
+            CreateCassetteOpData(**kwargs)
+
+    @pytest.mark.parametrize(
+        "kwargs",
+        [
+            {"position": 2, "op_type": "refill", "bills": 1, "count": 5},
+            {"position": 3, "op_type": "empty", "bills": 1},
+            {"position": 1, "op_type": "recount", "count": 1, "denomination": 50},
+        ],
+    )
+    def test_rejects_a_type_carrying_a_foreign_field(self, kwargs):
+        """An op that carries two meanings is ambiguous on the wire, and the
+        machine would have to guess which one to apply."""
+        with pytest.raises(ValidationError):
+            CreateCassetteOpData(**kwargs)
+
+    def test_rejects_a_refill_of_zero_or_fewer_notes(self):
+        """A refill is a delta that adds notes. Zero is a no-op an operator
+        did not mean, and negative is a withdrawal wearing a refill's name."""
+        for bills in (0, -5):
+            with pytest.raises(ValidationError):
+                CreateCassetteOpData(position=2, op_type="refill", bills=bills)
+
+    def test_allows_a_recount_to_zero(self):
+        """Distinct from refill: counting a bay and finding it empty is a real
+        and important observation."""
+        assert CreateCassetteOpData(position=2, op_type="recount", count=0).count == 0
+
+    def test_rejects_a_negative_recount_and_a_non_positive_denomination(self):
+        with pytest.raises(ValidationError):
+            CreateCassetteOpData(position=2, op_type="recount", count=-1)
+        with pytest.raises(ValidationError):
+            CreateCassetteOpData(position=2, op_type="set_denomination", denomination=0)
+
+    def test_rejects_an_unknown_type_and_a_non_positive_position(self):
+        with pytest.raises(ValidationError):
+            CreateCassetteOpData(position=1, op_type="drain")
+        with pytest.raises(ValidationError):
+            CreateCassetteOpData(position=0, op_type="empty")
+
+
+class TestWireShape:
+    def test_each_type_ships_only_its_own_field(self):
+        assert op(op_type="refill", bills=100).to_wire_dict() == {
+            "id": "op-1",
+            "at": 1790106060,
+            "type": "refill",
+            "position": 2,
+            "bills": 100,
+        }
+        assert op(op_type="empty").to_wire_dict() == {
+            "id": "op-1",
+            "at": 1790106060,
+            "type": "empty",
+            "position": 2,
+        }
+        assert op(op_type="recount", count=37).to_wire_dict()["count"] == 37
+        assert (
+            op(op_type="set_denomination", denomination=50).to_wire_dict()[
+                "denomination"
+            ]
+            == 50
+        )
+
+    def test_nulls_never_reach_the_wire(self):
+        """The row has three nullable columns and one op only ever means one
+        of them. Shipping the other two as null would make the machine guess."""
+        for op_type in CASSETTE_OP_TYPES:
+            kw = {
+                "refill": {"bills": 1},
+                "recount": {"count": 1},
+                "set_denomination": {"denomination": 1},
+                "empty": {},
+            }[op_type]
+            wire = op(op_type=op_type, **kw).to_wire_dict()
+            assert None not in wire.values()
+
+    def test_payload_declares_v2_and_preserves_order(self):
+        ops = [
+            op(id="a", op_type="refill", bills=1),
+            op(id="b", op_type="empty"),
+        ]
+        wire = PublishCassetteOpsPayload(ops=ops).to_wire_dict()
+        assert wire["schema_version"] == 2
+        assert [o["id"] for o in wire["ops"]] == ["a", "b"]
+
+    def test_an_empty_window_is_representable(self):
+        """A machine with no operator history still gets a well-formed
+        payload rather than the publisher having to special-case it."""
+        assert PublishCassetteOpsPayload(ops=[]).to_wire_dict() == {
+            "schema_version": 2,
+            "ops": [],
+        }