test: dispense outcome capture — handler transitions, payment gate, resume op, hold mirror

Twenty tests in the project's style (asyncio.run, monkeypatched crud, no
DB): confirmed → pending + distribution; nothing out → cash_owed; some
out → partial_pending with nothing spawned; already-captured recorded
not moved; identical resend acked without a row; report-before-payment
stored unlinked and adopted when the payment lands; remediation moves
the owed settlement; a remediation that did not confirm leaves it;
unpaired sender / malformed body refused. The payment gate: cash_out →
awaiting_dispense with no distribution, cash_in unchanged. The
resume_cash_out op's position rule and wire shape, the worklist model's
new buckets, and the state-document hold mirror set-and-clear. The
existing nulls-never-reach-the-wire test learns the machine-wide op.

tests/test_cassette_ops.py

@@ -124,6 +124,8 @@ class TestWireShape:
                 "recount": {"count": 1},
                 "set_denomination": {"denomination": 1},
                 "empty": {},
+                # machine-wide (ADR-005 §5): position 0, no position on the wire
+                "resume_cash_out": {"position": 0},
             }[op_type]
             wire = op(op_type=op_type, **kw).to_wire_dict()
             assert None not in wire.values()

tests/test_dispense_outcome.py

@@ -0,0 +1,542 @@
+"""
+Tests for dispense-outcome capture (bitspire ADR-005 §1-§2, #122).
+
+Covers the pure pieces and the handler's transitions with the crud layer
+monkeypatched (no DB), in the project's established style: asyncio.run inside
+the test body, SimpleNamespace for request/payment shapes.
+
+  - models: DispenseReportIn validation + derived numbers; resume_cash_out as a
+    machine-wide op (position 0, no position on the wire); the three new
+    worklist buckets default empty.
+  - handler: confirmed → pending + distribution spawned; nothing out →
+    cash_owed; some out → partial_pending; already-captured settlements are
+    recorded but not moved; a byte-identical resend is acked without a new
+    row; an orphan report (payment not landed) is stored unlinked; a
+    remediation report moves the owed settlement to pending; unpaired sender
+    and bad bodies are refused.
+  - gate: _handle_payment inserts cash_out as awaiting_dispense and does NOT
+    spawn distribution; cash_in is unchanged; an early report is adopted.
+  - consumer: the state document's cash_out_held_* mirrors onto the machine,
+    including clearing it.
+"""
+
+import asyncio
+from datetime import datetime, timezone
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+from pydantic import ValidationError
+
+from .. import crud as crud_mod
+from .. import dispense_transport, tasks
+from ..dispense_transport import (
+    _outcome_status,
+    handle_report_dispense,
+)
+from ..models import (
+    CASSETTE_OP_TYPES,
+    CassetteOp,
+    CreateCassetteOpData,
+    CreateDcaSettlementData,
+    DcaSettlement,
+    DispenseReportIn,
+    Machine,
+    PublishCassettesPayload,
+    StuckSettlementsResponse,
+)
+
+_NOW = datetime(2026, 10, 9, 7, 2, 33)
+_ATM_HEX = "df2003343784b69cb813b2a4fd231f83ae81133279251c735414f9909baa7ac6"
+_TXID = "tx_mv0madw6_wdhtea1v"
+_HASH = "6f216df32c36" + "0" * 52
+
+
+def _machine(**over) -> Machine:
+    base: dict[str, Any] = {
+        "id": "m1",
+        "operator_user_id": "op1",
+        "machine_npub": _ATM_HEX,
+        "wallet_id": "w1",
+        "name": "sintra",
+        "location": None,
+        "fiat_code": "EUR",
+        "is_active": True,
+        "created_at": _NOW,
+        "updated_at": _NOW,
+    }
+    base.update(over)
+    return Machine(**base)
+
+
+def _settlement(status="awaiting_dispense", **over) -> DcaSettlement:
+    base: dict[str, Any] = {
+        "id": "s1",
+        "machine_id": "m1",
+        "payment_hash": _HASH,
+        "bitspire_event_id": None,
+        "bitspire_txid": _TXID,
+        "wire_sats": 54440,
+        "fiat_amount": 40.0,
+        "fiat_code": "EUR",
+        "exchange_rate": 1361.0,
+        "principal_sats": 54440,
+        "fee_sats": 0,
+        "platform_fee_sats": 0,
+        "operator_fee_sats": 0,
+        "tx_type": "cash_out",
+        "bills_json": None,
+        "cassettes_json": None,
+        "status": status,
+        "error_message": None,
+        "processed_at": None,
+        "created_at": _NOW,
+    }
+    base.update(over)
+    return DcaSettlement(**base)
+
+
+def _report(**over) -> dict:
+    """The wire body for sintra's 2026-10-09 jam, as a dict."""
+    body: dict[str, Any] = {
+        "txid": _TXID,
+        "payment_hash": _HASH,
+        "tx_type": "cash_out",
+        "dispense_confirmed": False,
+        "error": "Note stopped at the cassette exit",
+        "error_code": "F56DispenseError",
+        "raw_code": "78 42",
+        "error_class": "terminal",
+        "fiat_cents": 4000,
+        "currency": "EUR",
+        "bills": [{"denomination": 20, "requested": 2, "dispensed": 0, "rejected": 0}],
+        "cassettes": [
+            {
+                "position": 1,
+                "denomination": 50,
+                "provisioned": 0,
+                "dispensed": 0,
+                "rejected": 0,
+            },
+            {
+                "position": 2,
+                "denomination": 20,
+                "provisioned": 2,
+                "dispensed": 0,
+                "rejected": 0,
+            },
+        ],
+        "counts_uncertain": True,
+        "at": 1791529353,
+    }
+    body.update(over)
+    return body
+
+
+def _confirmed() -> dict:
+    return _report(
+        dispense_confirmed=True,
+        error=None,
+        error_code=None,
+        raw_code=None,
+        error_class=None,
+        bills=[{"denomination": 20, "requested": 2, "dispensed": 2, "rejected": 0}],
+        counts_uncertain=False,
+    )
+
+
+def _partial() -> dict:
+    return _report(
+        bills=[{"denomination": 20, "requested": 2, "dispensed": 1, "rejected": 1}],
+        counts_uncertain=False,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Models
+# ---------------------------------------------------------------------------
+
+
+class TestDispenseReportIn:
+    def test_derived_numbers(self):
+        r = DispenseReportIn(**_partial())
+        assert r.dispensed_fiat_cents == 2000
+        assert r.total_dispensed_notes == 1
+        assert DispenseReportIn(**_confirmed()).dispensed_fiat_cents == 4000
+        assert DispenseReportIn(**_report()).total_dispensed_notes == 0
+
+    def test_outcome_routing(self):
+        assert _outcome_status(DispenseReportIn(**_confirmed())) == "pending"
+        assert _outcome_status(DispenseReportIn(**_partial())) == "partial_pending"
+        assert _outcome_status(DispenseReportIn(**_report())) == "cash_owed"
+
+    def test_rejects_cash_in_and_unknown_class(self):
+        with pytest.raises(ValidationError):
+            DispenseReportIn(**_report(tx_type="cash_in"))
+        with pytest.raises(ValidationError):
+            DispenseReportIn(**_report(error_class="weird"))
+        with pytest.raises(ValidationError):
+            DispenseReportIn(**_report(txid="  "))
+        with pytest.raises(ValidationError):
+            DispenseReportIn(**_report(fiat_cents=-1))
+
+
+class TestResumeCashOutOp:
+    def test_is_a_known_type_and_machine_wide(self):
+        assert "resume_cash_out" in CASSETTE_OP_TYPES
+        op = CassetteOp(
+            id="r1",
+            machine_id="m1",
+            position=0,
+            op_type="resume_cash_out",
+            created_at=_NOW,
+        )
+        wire = op.to_wire_dict()
+        assert wire == {
+            "id": "r1",
+            "at": int(_NOW.timestamp()),
+            "type": "resume_cash_out",
+        }
+        assert "position" not in wire
+
+    def test_position_must_be_zero_for_resume_and_positive_otherwise(self):
+        with pytest.raises(ValidationError):
+            CassetteOp(
+                id="r1",
+                machine_id="m1",
+                position=2,
+                op_type="resume_cash_out",
+                created_at=_NOW,
+            )
+        with pytest.raises(ValidationError):
+            CreateCassetteOpData(position=1, op_type="resume_cash_out")
+        with pytest.raises(ValidationError):
+            CreateCassetteOpData(position=0, op_type="refill", bills=5)
+        CreateCassetteOpData(position=0, op_type="resume_cash_out")  # ok
+
+    def test_resume_carries_no_bay_fields(self):
+        with pytest.raises(ValidationError):
+            CreateCassetteOpData(position=0, op_type="resume_cash_out", count=3)
+
+
+class TestWorklistModel:
+    def test_new_buckets_default_empty(self):
+        r = StuckSettlementsResponse(
+            threshold_minutes=30,
+            rejected=[],
+            errored=[],
+            stuck_pending=[],
+            stuck_processing=[],
+        )
+        assert (
+            r.cash_owed == []
+            and r.partial_pending == []
+            and r.dispense_unreported == []
+        )
+
+
+# ---------------------------------------------------------------------------
+# Handler
+# ---------------------------------------------------------------------------
+
+
+class _Wired:
+    """Monkeypatched crud layer for handle_report_dispense."""
+
+    def __init__(self, monkeypatch, *, machine, settlement, existing_report=None):
+        self.inserted = []
+        self.applied = []
+        self.spawned = []
+        self.uncertain = []
+        self.statuses = []
+        self.settlement = settlement
+
+        async def get_machine(_hex):
+            return machine
+
+        async def get_report(_mid, _txid, _at):
+            return existing_report
+
+        async def get_settlement(_mid, txid):
+            if self.settlement is not None and self.settlement.bitspire_txid == txid:
+                return self.settlement
+            return None
+
+        async def insert(mid, sid, report):
+            self.inserted.append((mid, sid, report))
+            return SimpleNamespace(id="rep1", settlement_id=sid, txid=report.txid)
+
+        async def apply(sid, report, new_status, reported_at):
+            self.applied.append((sid, new_status, reported_at))
+            return (
+                self.settlement.copy(update={"status": new_status})
+                if self.settlement
+                else None
+            )
+
+        async def set_uncertain(mid, since):
+            self.uncertain.append((mid, since))
+
+        async def mark_status(sid, status, _msg):
+            self.statuses.append((sid, status))
+            return None
+
+        monkeypatch.setattr(
+            dispense_transport, "get_machine_by_atm_pubkey_hex", get_machine
+        )
+        monkeypatch.setattr(dispense_transport, "get_dispense_report", get_report)
+        monkeypatch.setattr(
+            dispense_transport, "get_settlement_by_txid", get_settlement
+        )
+        monkeypatch.setattr(dispense_transport, "insert_dispense_report", insert)
+        monkeypatch.setattr(dispense_transport, "apply_dispense_outcome", apply)
+        monkeypatch.setattr(
+            dispense_transport, "set_machine_counts_uncertain", set_uncertain
+        )
+        monkeypatch.setattr(
+            dispense_transport, "_spawn_distribution", self.spawned.append
+        )
+        monkeypatch.setattr(crud_mod, "mark_settlement_status", mark_status)
+
+
+def _req(body, sender=_ATM_HEX):
+    return SimpleNamespace(body=body, sender_pubkey=sender, event_id="ev1")
+
+
+class TestHandleReportDispense:
+    def test_confirmed_captures_and_distributes(self, monkeypatch):
+        w = _Wired(monkeypatch, machine=_machine(), settlement=_settlement())
+        out = asyncio.run(handle_report_dispense(None, _req(_confirmed())))
+        assert out["received"] is True
+        assert out["settlement_status"] == "pending"
+        assert w.applied == [("s1", "pending", datetime.fromtimestamp(1791529353))]
+        assert w.spawned == ["s1"]
+        assert w.inserted[0][1] == "s1"  # linked to the settlement
+        assert w.uncertain == []
+
+    def test_nothing_out_is_cash_owed_and_nothing_moves(self, monkeypatch):
+        w = _Wired(monkeypatch, machine=_machine(), settlement=_settlement())
+        out = asyncio.run(handle_report_dispense(None, _req(_report())))
+        assert out["settlement_status"] == "cash_owed"
+        assert w.applied[0][1] == "cash_owed"
+        assert w.spawned == []
+        # counts_uncertain on the report mirrors onto the machine immediately
+        assert len(w.uncertain) == 1 and w.uncertain[0][0] == "m1"
+
+    def test_some_out_is_partial_pending_held_whole(self, monkeypatch):
+        w = _Wired(monkeypatch, machine=_machine(), settlement=_settlement())
+        out = asyncio.run(handle_report_dispense(None, _req(_partial())))
+        assert out["settlement_status"] == "partial_pending"
+        assert w.spawned == []  # ADR-005 Decision 1: one distribution, when final
+
+    def test_already_captured_settlement_is_recorded_not_moved(self, monkeypatch):
+        w = _Wired(
+            monkeypatch, machine=_machine(), settlement=_settlement(status="processed")
+        )
+        out = asyncio.run(handle_report_dispense(None, _req(_report())))
+        assert out["settlement_status"] == "processed"
+        assert w.applied == [] and w.spawned == []
+        assert len(w.inserted) == 1  # the row still lands — it is information
+
+    def test_identical_resend_is_acked_without_a_new_row(self, monkeypatch):
+        existing = SimpleNamespace(id="rep0", settlement_id="s1", txid=_TXID)
+        w = _Wired(
+            monkeypatch,
+            machine=_machine(),
+            settlement=_settlement(status="cash_owed"),
+            existing_report=existing,
+        )
+        out = asyncio.run(handle_report_dispense(None, _req(_report())))
+        assert out["received"] is True and out.get("duplicate") is True
+        assert out["settlement_status"] == "cash_owed"
+        assert w.inserted == [] and w.applied == []
+
+    def test_report_before_payment_is_stored_unlinked(self, monkeypatch):
+        w = _Wired(monkeypatch, machine=_machine(), settlement=None)
+        out = asyncio.run(handle_report_dispense(None, _req(_confirmed())))
+        assert out["settlement_status"] is None
+        assert w.inserted[0][1] is None
+        assert w.spawned == []
+
+    def test_remediation_moves_the_owed_settlement_to_pending(self, monkeypatch):
+        owed = _settlement(status="cash_owed")
+        w = _Wired(monkeypatch, machine=_machine(), settlement=owed)
+        body = _confirmed()
+        body.update(txid="manual-1", remediates_txid=_TXID, at=1791530000)
+        out = asyncio.run(handle_report_dispense(None, _req(body)))
+        assert out["settlement_status"] == "pending"
+        assert w.statuses == [("s1", "pending")]
+        assert w.spawned == ["s1"]
+        assert w.applied == []  # the ORIGINAL report's columns stay on the settlement
+
+    def test_remediation_that_did_not_confirm_leaves_it_owed(self, monkeypatch):
+        w = _Wired(
+            monkeypatch, machine=_machine(), settlement=_settlement(status="cash_owed")
+        )
+        body = _report()
+        body.update(txid="manual-2", remediates_txid=_TXID, at=1791530001)
+        out = asyncio.run(handle_report_dispense(None, _req(body)))
+        assert out["settlement_status"] == "cash_owed"
+        assert w.statuses == [] and w.spawned == []
+
+    def test_unpaired_sender_and_bad_body_are_refused(self, monkeypatch):
+        _Wired(monkeypatch, machine=None, settlement=None)
+        with pytest.raises(ValueError, match="not a paired machine"):
+            asyncio.run(handle_report_dispense(None, _req(_report())))
+        _Wired(monkeypatch, machine=_machine(), settlement=_settlement())
+        with pytest.raises(ValueError, match="invalid report_dispense body"):
+            asyncio.run(handle_report_dispense(None, _req({"txid": _TXID})))
+        with pytest.raises(ValueError, match="sender_pubkey"):
+            asyncio.run(handle_report_dispense(None, _req(_report(), sender="")))
+
+
+# ---------------------------------------------------------------------------
+# Gate in _handle_payment
+# ---------------------------------------------------------------------------
+
+
+def _payment(is_in=True):
+    return SimpleNamespace(
+        success=True,
+        wallet_id="w1",
+        extra={
+            "source": "bitspire",
+            "type": "cash_out" if is_in else "cash_in",
+            "txid": _TXID,
+        },
+        is_in=is_in,
+        sat=54440 if is_in else -54440,
+        payment_hash=_HASH,
+    )
+
+
+def _data(tx_type) -> CreateDcaSettlementData:
+    return CreateDcaSettlementData(
+        machine_id="m1",
+        payment_hash=_HASH,
+        bitspire_txid=_TXID,
+        wire_sats=54440,
+        fiat_amount=40.0,
+        fiat_code="EUR",
+        exchange_rate=1361.0,
+        principal_sats=54440,
+        fee_sats=0,
+        platform_fee_sats=0,
+        operator_fee_sats=0,
+        tx_type=tx_type,
+    )
+
+
+class _GateWired:
+    def __init__(self, monkeypatch, *, tx_type, early_report=None):
+        self.created = []
+        self.spawned = []
+        self.adopted = []
+
+        async def get_machine(_wid):
+            return _machine()
+
+        def attribution(_machine, _extra):
+            return None
+
+        async def get_super():
+            return SimpleNamespace(id="default")
+
+        def parse(**_kw):
+            return _data(tx_type)
+
+        async def create(data, initial_status, error_message=None):
+            self.created.append((data.tx_type, initial_status))
+            return _settlement(status=initial_status, tx_type=data.tx_type)
+
+        async def process(sid):
+            self.spawned.append(sid)
+
+        async def early(_mid, _txid):
+            return early_report
+
+        async def adopt(settlement, machine, row):
+            self.adopted.append((settlement.id, row.id))
+            return "pending"
+
+        monkeypatch.setattr(tasks, "get_active_machine_by_wallet_id", get_machine)
+        monkeypatch.setattr(tasks, "assert_nostr_attribution", attribution)
+        monkeypatch.setattr(tasks, "get_super_config", get_super)
+        monkeypatch.setattr(tasks, "parse_settlement", parse)
+        monkeypatch.setattr(tasks, "create_settlement_idempotent", create)
+        monkeypatch.setattr(tasks, "process_settlement", process)
+        monkeypatch.setattr(crud_mod, "get_latest_unlinked_dispense_report", early)
+        monkeypatch.setattr(dispense_transport, "adopt_unlinked_report", adopt)
+
+
+async def _drain():
+    # let any create_task'd distribution run
+    await asyncio.sleep(0)
+
+
+class TestPaymentGate:
+    def test_cash_out_lands_awaiting_dispense_and_does_not_distribute(
+        self, monkeypatch
+    ):
+        w = _GateWired(monkeypatch, tx_type="cash_out")
+
+        async def run():
+            await tasks._handle_payment(_payment(is_in=True))
+            await _drain()
+
+        asyncio.run(run())
+        assert w.created == [("cash_out", "awaiting_dispense")]
+        assert w.spawned == []
+        assert w.adopted == []
+
+    def test_cash_in_is_unchanged(self, monkeypatch):
+        w = _GateWired(monkeypatch, tx_type="cash_in")
+
+        async def run():
+            await tasks._handle_payment(_payment(is_in=False))
+            await _drain()
+
+        asyncio.run(run())
+        assert w.created == [("cash_in", "pending")]
+        assert w.spawned == ["s1"]
+
+    def test_early_report_is_adopted_when_the_payment_lands(self, monkeypatch):
+        early = SimpleNamespace(id="rep-early", txid=_TXID)
+        w = _GateWired(monkeypatch, tx_type="cash_out", early_report=early)
+
+        async def run():
+            await tasks._handle_payment(_payment(is_in=True))
+            await _drain()
+
+        asyncio.run(run())
+        assert w.adopted == [("s1", "rep-early")]
+        assert w.spawned == []  # adoption decides; the fake adopt did not spawn
+
+
+# ---------------------------------------------------------------------------
+# Consumer mirror
+# ---------------------------------------------------------------------------
+
+
+class TestCashOutHoldMirror:
+    def test_sets_and_clears_from_the_state_document(self):
+        calls = []
+
+        async def setter(mid, since, reason, code):
+            calls.append((mid, since, reason, code))
+
+        held = PublishCassettesPayload(
+            positions={"2": {"denomination": 20, "count": 54}},
+            cash_out_held_since=1791529353,
+            cash_out_held_reason="Note stopped at the cassette exit",
+            cash_out_held_code="78 42",
+        )
+        clear = PublishCassettesPayload(
+            positions={"2": {"denomination": 20, "count": 54}}
+        )
+        asyncio.run(tasks._record_cash_out_hold("m1", held, setter))
+        asyncio.run(tasks._record_cash_out_hold("m1", clear, setter))
+        assert calls[0][0] == "m1"
+        assert calls[0][1] == datetime.fromtimestamp(1791529353, tz=timezone.utc)
+        assert calls[0][2:] == ("Note stopped at the cassette exit", "78 42")
+        assert calls[1] == ("m1", None, None, None)