feat(cassettes): persist the machine's counts-uncertain marker

The machine has been publishing counts_uncertain_since since v1 of the
state document and spirekeeper has been parsing it into a field nobody
read. That defeats the point of the marker: it exists so a human opens
the bay and recounts.

A dispenser can throw, or time out, after notes have physically moved.
The machine cannot know how many left, so rather than decrement a number
it would be guessing at, it stamps the moment (bitspire ADR-004,
decision 3). m014 gives that stamp a home on dca_machines, and the
consumer mirrors it on every state event.

Stored on the machine rather than the bay because the uncertainty is
about the dispense as a whole; a multi-bay dispense that fails midway
gives no reliable way to attribute it to one position.

Written through even when the machine reports None. The machine clearing
the marker is as important as setting it — the operator recounted, the
bay is trustworthy again — and a banner that never goes away is a banner
nobody reads.

crud.py

@@ -257,6 +257,25 @@ async def set_machine_unpaired(machine_id: str) -> Machine | None:
     return await get_machine(machine_id)
 
 
+async def set_machine_counts_uncertain(machine_id: str, since: datetime | None) -> None:
+    """Record (or clear) the machine's own "I can't vouch for these counts"
+    marker, straight from its state document.
+
+    `updated_at` is deliberately left alone. This is the machine reporting
+    about itself on a five-minute heartbeat, not an operator editing the
+    machine, and touching the timestamp on every heartbeat would make the
+    registry look perpetually just-modified.
+    """
+    await db.execute(
+        """
+        UPDATE spirekeeper.dca_machines
+           SET counts_uncertain_since = :since
+         WHERE id = :id
+        """,
+        {"since": since, "id": machine_id},
+    )
+
+
 async def delete_machine(machine_id: str) -> None:
     await db.execute(
         "DELETE FROM spirekeeper.dca_machines WHERE id = :id",

migrations.py

@@ -901,3 +901,24 @@ async def m013_add_cassette_ops(db):
         "CREATE INDEX IF NOT EXISTS cassette_ops_machine_idx "
         "ON cassette_ops (machine_id, created_at DESC)"
     )
+
+
+async def m014_add_counts_uncertain_since(db):
+    """Surface the machine's "I don't know what left the bay" marker.
+
+    A dispenser can throw, or time out, after notes have physically moved. The
+    machine cannot know how many left, so rather than decrement a number it
+    would be guessing at, it stamps the moment and reports it (bitspire
+    ADR-004, decision 3). It has been publishing this field since v1 of the
+    state document and the operator has been discarding it, which defeats the
+    point: the marker exists to tell a human to open the bay and recount.
+
+    Stored on the machine, not the bay, because the uncertainty is about the
+    dispense as a whole — a multi-bay dispense that fails midway leaves no
+    reliable way to attribute it to one position. Cleared to NULL by the
+    machine's own report once it is confident again.
+    """
+    await db.execute(
+        "ALTER TABLE spirekeeper.dca_machines "
+        "ADD COLUMN counts_uncertain_since TIMESTAMP"
+    )

models.py

@@ -63,6 +63,10 @@ class Machine(BaseModel):
     # NIP-46 bunker pairing (S0 / #9). NULL until the spire is first paired.
     bunker_spire_key_name: str | None = None
     paired_at: datetime | None = None
+    # Set when the machine reports that a dispense ended without a reliable
+    # count of what left the bay; cleared by the machine's own report. The
+    # dashboard turns this into a prompt to open the bay and recount.
+    counts_uncertain_since: datetime | None = None
     created_at: datetime
     updated_at: datetime
 

tasks.py

@@ -339,6 +339,7 @@ async def _cassette_consumer_tick(current_filter_key: str | None) -> str:
         get_machine_by_atm_pubkey_hex,
         list_all_active_machines,
         mark_cassette_ops_acked,
+        set_machine_counts_uncertain,
     )
 
     machines = await list_all_active_machines()
@@ -375,6 +376,7 @@ async def _cassette_consumer_tick(current_filter_key: str | None) -> str:
                     get_machine_by_atm_pubkey_hex,
                     apply_reported_state,
                     mark_cassette_ops_acked,
+                    set_machine_counts_uncertain,
                 )
             except Exception as exc:
                 logger.warning(
@@ -410,11 +412,38 @@ async def _record_op_acknowledgements(
         )
 
 
+async def _record_counts_uncertainty(
+    machine_id: str, payload, set_machine_counts_uncertain
+) -> None:
+    """Mirror the machine's counts-uncertain marker onto its registry row.
+
+    The machine sets this when a dispense ended without a reliable count of
+    what physically left the bay — a dispenser throw, or a timeout. It cannot
+    know how many notes moved, so it says so instead of decrementing a number
+    it would be guessing at.
+
+    Written on every state event, including when it is None, because the
+    machine clearing the marker is exactly as important as setting it: the
+    operator has recounted, the bay is trustworthy again, and a banner that
+    never goes away is a banner nobody reads.
+    """
+    from datetime import datetime as _datetime
+    from datetime import timezone as _timezone
+
+    since = None
+    if payload.counts_uncertain_since is not None:
+        since = _datetime.fromtimestamp(
+            int(payload.counts_uncertain_since), tz=_timezone.utc
+        )
+    await set_machine_counts_uncertain(machine_id, since)
+
+
 async def _handle_cassette_state_event(
     event_message,
     get_machine_by_atm_pubkey_hex,
     apply_reported_state,
     mark_cassette_ops_acked,
+    set_machine_counts_uncertain,
 ) -> None:
     """Verify signature, resolve the operator's signer, decrypt via the
     signer abstraction (bunker round-trip for RemoteBunkerSigner; direct
@@ -526,5 +555,6 @@ async def _handle_cassette_state_event(
         )
 
     # Acknowledgement runs regardless of whether the counts were newer — see
-    # _record_op_acknowledgements for why.
+    # _record_op_acknowledgements for why. Same for the uncertainty marker.
     await _record_op_acknowledgements(machine.id, payload, mark_cassette_ops_acked)
+    await _record_counts_uncertainty(machine.id, payload, set_machine_counts_uncertain)

tests/test_cassette_ops.py

@@ -317,3 +317,40 @@ class TestRecordOpAcknowledgements:
             "m1", state_payload(applied_ops=["late-ack"]), mark
         )
         assert seen == ["late-ack"]
+
+
+# =============================================================================
+# _record_counts_uncertainty — the machine saying "don't trust these counts"
+# =============================================================================
+
+
+class TestRecordCountsUncertainty:
+    @pytest.mark.asyncio
+    async def test_stores_the_reported_moment_as_utc(self):
+        calls = []
+
+        async def setter(machine_id, since):
+            calls.append((machine_id, since))
+
+        await tasks._record_counts_uncertainty(
+            "m1", state_payload(counts_uncertain_since=1790110546), setter
+        )
+        assert len(calls) == 1
+        machine_id, since = calls[0]
+        assert machine_id == "m1"
+        assert since is not None
+        assert since.tzinfo is not None
+        assert int(since.timestamp()) == 1790110546
+
+    @pytest.mark.asyncio
+    async def test_clears_the_marker_when_the_machine_is_confident_again(self):
+        """A banner that never goes away is a banner nobody reads. The
+        machine dropping the field is how the operator learns the recount
+        took, so None must be written through rather than skipped."""
+        calls = []
+
+        async def setter(machine_id, since):
+            calls.append((machine_id, since))
+
+        await tasks._record_counts_uncertainty("m1", state_payload(), setter)
+        assert calls == [("m1", None)]