From d8190375a6650eb3852899f144dcae6c6442c4ba Mon Sep 17 00:00:00 2001 From: Padreug Date: Wed, 23 Sep 2026 09:50:28 +0200 Subject: [PATCH] feat(cassettes): schema and models for operator operations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- migrations.py | 41 +++++++++ models.py | 166 ++++++++++++++++++++++++++++++++++++- tests/test_cassette_ops.py | 142 +++++++++++++++++++++++++++++++ 3 files changed, 348 insertions(+), 1 deletion(-) create mode 100644 tests/test_cassette_ops.py 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": "", "at": 1790106060, "type": "refill", +# "position": 2, "bills": 100}, +# {"id": "", "at": 1790106061, "type": "empty", "position": 3}, +# {"id": "", "at": 1790106062, "type": "recount", +# "position": 1, "count": 37}, +# {"id": "", "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": [], + }