plm-lottery/tests/unit/test_scheduler.py

from datetime import datetime, timedelta, timezone

import pytest
from embit.transaction import Transaction
from sqlalchemy import select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from app.config import settings
from app.db.base import Base
from tests.unit.fake_network import FakeNetwork
from app.db.models import AuditLog, PendingTransaction, Round, RoundConfig, RoundParticipant, User
from app.rounds.scheduler import RoundScheduler, _reserved_payout_outpoints
from app.wallet.hd import derive_pool_address
from app.wallet.psbt_builder import MAX_TX_INPUTS


class FakeListener:
    client = object()  # truthy sentinel; _tick only checks "is not None"
    tip_height = 100
    tip_header_hex = "00"


@pytest.fixture
async def session_factory():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield async_sessionmaker(engine, expire_on_commit=False)
    await engine.dispose()


async def test_tick_survives_sqlite_naive_datetime_roundtrip(session_factory, monkeypatch):
    """Regression test: SQLite drops tzinfo on round-trip, so opened_at comes back
    naive even though it was written as an aware UTC datetime. A prior bug compared
    it directly against datetime.now(timezone.utc) and crashed with
    "can't compare offset-naive and offset-aware datetimes" on every tick once a
    round existed — this must not happen."""
    async with session_factory() as session:
        session.add(RoundConfig(fee_address="", round_duration_seconds=3600))  # not due yet
        session.add(Round(status="open", opened_at=datetime.now(timezone.utc)))
        await session.commit()

    scheduler = RoundScheduler(session_factory, FakeListener())
    await scheduler._tick()  # must not raise


async def test_tick_closes_round_with_no_participants_once_due(session_factory, monkeypatch):
    past = datetime.now(timezone.utc) - timedelta(seconds=10)
    async with session_factory() as session:
        session.add(RoundConfig(fee_address="", round_duration_seconds=1))
        # B-61: the deadline comes from the round's own snapshot, not from the config.
        session.add(Round(status="open", opened_at=past, duration_seconds=1))
        await session.commit()

    scheduler = RoundScheduler(session_factory, FakeListener())
    await scheduler._tick()

    async with session_factory() as session:
        round_ = (await session.scalars(select(Round))).one()
        assert round_.status == "closed"


# --- B-25: the payout must be persisted before it is broadcast, like bets/withdrawals ---

# A real, reusable PLM bech32 address so build_payout_transaction's
# script.Script.from_address(...) succeeds — this is not a value the scheduler
# validates itself (that's the admin panel's job for fee_address), it just needs to
# actually decode.
_WINNER_ADDRESS = "plm1q0xcqpzrky6eff2g52qdye53xkk9jxkvraxkkwd"
_FEE_ADDRESS = "plm1q5x25wd6q463mfhckjraaedgjg0lyu73qfcj43n"

_POOL_AMOUNT_SATS = 10_000_000_000  # 100 PLM


class FakePayoutClient:
    def __init__(self, entries, *, fail_broadcast=False):
        self._entries = entries
        self._fail_broadcast = fail_broadcast
        self.broadcasted: list[str] = []

    async def listunspent(self, scripthash):
        return self._entries

    async def broadcast(self, raw_tx_hex):
        if self._fail_broadcast:
            raise RuntimeError("node rejected the transaction")
        self.broadcasted.append(raw_tx_hex)
        return "network-txid"


class FakePayoutListener(FakeNetwork):
    """The listener as the scheduler sees it: the active `client`, plus the network
    surface a broadcast verdict uses (B-76)."""

    def __init__(self, client, known: bool | None = False):
        super().__init__(known=known)
        self.client = client


@pytest.fixture
async def payout_session_factory(tmp_path, monkeypatch):
    """Same master-key bootstrap as test_broadcast.py's fixture: _trigger_payout
    needs a real pool key to sign with."""
    monkeypatch.setattr(settings, "master_key_path", str(tmp_path / "master.xprv.enc"))
    monkeypatch.setattr(
        settings,
        "xprv_encryption_key",
        __import__("cryptography.fernet", fromlist=["Fernet"]).Fernet.generate_key().decode(),
    )
    from app.wallet import hd

    hd._account_key = None
    hd.generate_master_key()

    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield async_sessionmaker(engine, expire_on_commit=False)
    await engine.dispose()
    hd._account_key = None


def _payout_bet_txid(index: int) -> str:
    return f"{index + 0x33:064x}"


async def _seed_paying_out_round(session_factory, round_id: int = 1, bets: list[int] | None = None) -> int:
    """A round in `paying_out` whose confirmed participants — the first one is the winner —
    each placed a bet of the given net amount (B-82: the payout spends exactly those bet
    outputs, `bet_txid:0`, so the pool is their sum)."""
    bets = bets if bets is not None else [_POOL_AMOUNT_SATS]
    async with session_factory() as session:
        winner = User(username="winner", password_hash="x", derivation_index=0, address=_WINNER_ADDRESS)
        session.add(winner)
        await session.flush()
        session.add(RoundConfig(fee_address=_FEE_ADDRESS, fee_rate_sat_vb=1))
        session.add(
            Round(
                id=round_id,
                status="paying_out",
                pool_amount_sats=sum(bets),
                winner_user_id=winner.id,
                pool_address=derive_pool_address(round_id),  # B-80
            )
        )
        await session.flush()
        for index, amount in enumerate(bets):
            user = winner
            if index:
                user = User(
                    username=f"bettor{index}", password_hash="x", derivation_index=index,
                    address=derive_pool_address(7000 + index),
                )
                session.add(user)
                await session.flush()
            session.add(
                RoundParticipant(
                    round_id=round_id, user_id=user.id, bet_amount_sats=amount, bet_txid=_payout_bet_txid(index),
                    bet_vout=0, status="confirmed", draw_position=index,
                )
            )
        await session.commit()
        return winner.id


async def test_trigger_payout_persists_before_broadcasting(payout_session_factory):
    """The happy path: payout_txid and a PendingTransaction must exist once the
    broadcast succeeds, promoted from "building" to "pending" — the two-phase write
    that used to be missing entirely (B-25)."""
    await _seed_paying_out_round(payout_session_factory)
    entries = [{"tx_hash": "33" * 32, "tx_pos": 0, "height": 10, "value": _POOL_AMOUNT_SATS + 100_000}]
    client = FakePayoutClient(entries)
    scheduler = RoundScheduler(payout_session_factory, FakePayoutListener(client))

    await scheduler._trigger_payout(1)

    assert client.broadcasted
    async with payout_session_factory() as session:
        round_ = await session.get(Round, 1)
        assert round_.payout_txid is not None
        assert round_.winner_amount_sats and round_.fee_amount_sats

        pending = (await session.scalars(select(PendingTransaction))).one()
        assert pending.kind == "payout"
        assert pending.status == "pending"
        assert pending.current_txid == round_.payout_txid

        events = [e.event_type for e in (await session.scalars(select(AuditLog))).all()]
        assert "payout_sent" in events


async def test_trigger_payout_always_carries_a_bump_reserve_to_the_reserve_address(payout_session_factory):
    """B-79 (RND-23, RND-24, RND-38, DAT-22, DAT-73): the pool holds exactly the round's bets
    and the payout spends them all, yet the transaction has three outputs — winner, fee
    address, reserve at m/84'/746'/0'/1/0 — and the round records the reserve before the
    broadcast. Nothing is created or lost to the satoshi."""
    from app.wallet.psbt_builder import estimate_vsize, payout_bump_reserve

    await _seed_paying_out_round(payout_session_factory)
    entries = [{"tx_hash": "34" * 32, "tx_pos": 0, "height": 10, "value": _POOL_AMOUNT_SATS}]  # no excess
    client = FakePayoutClient(entries)
    scheduler = RoundScheduler(payout_session_factory, FakePayoutListener(client))

    await scheduler._trigger_payout(1)

    async with payout_session_factory() as session:
        round_ = await session.get(Round, 1)
        pending = (await session.scalars(select(PendingTransaction))).one()
    tx = Transaction.parse(bytes.fromhex(pending.raw_tx_hex))
    from app.wallet.plm_network import PLM_MAINNET

    addresses = [o.script_pubkey.address(network=PLM_MAINNET) for o in tx.vout]
    assert addresses == [_WINNER_ADDRESS, _FEE_ADDRESS, derive_pool_address()]  # reserve = 1/0
    reserve = payout_bump_reserve(1)
    network_fee = estimate_vsize(1, 3)
    winner_share = _POOL_AMOUNT_SATS * 70 // 100
    assert round_.bump_reserve_sats == reserve == tx.vout[2].value
    assert round_.winner_amount_sats == winner_share - network_fee == tx.vout[0].value
    assert round_.fee_amount_sats == _POOL_AMOUNT_SATS - winner_share - reserve == tx.vout[1].value
    assert (
        round_.winner_amount_sats + round_.fee_amount_sats + round_.bump_reserve_sats + network_fee
        == _POOL_AMOUNT_SATS
    )


async def test_trigger_payout_pays_a_round_with_more_participants_than_max_tx_inputs(
    payout_session_factory,
):  # B-52
    """End-to-end shape of the deadlock this fixes: the pool holds one UTXO per bet,
    so a round past MAX_TX_INPUTS participants could not be paid at all — the build
    failed with too_many_inputs, the round stayed "paying_out" retrying every 60s,
    and no new round could ever open behind it. It must now broadcast normally."""
    participants = MAX_TX_INPUTS + 1
    bet_sats = _POOL_AMOUNT_SATS // participants
    bets = [bet_sats] * participants
    # The pool's total is the round's recorded pool_amount_sats: integer division above
    # leaves a remainder, so top the last one up.
    bets[-1] += _POOL_AMOUNT_SATS - bet_sats * participants
    await _seed_paying_out_round(payout_session_factory, bets=bets)
    client = FakePayoutClient([])
    scheduler = RoundScheduler(payout_session_factory, FakePayoutListener(client))

    await scheduler._trigger_payout(1)

    assert client.broadcasted
    async with payout_session_factory() as session:
        pending = (await session.scalars(select(PendingTransaction))).one()
        assert pending.status == "pending"
        assert len(Transaction.parse(bytes.fromhex(pending.raw_tx_hex)).vin) == participants

        events = [e.event_type for e in (await session.scalars(select(AuditLog))).all()]
        assert "payout_sent" in events
        assert "payout_failed" not in events


async def test_trigger_payout_broadcast_failure_leaves_a_recoverable_row(payout_session_factory):
    """Before B-25, a broadcast rejection here left nothing behind — no payout_txid,
    no PendingTransaction — because everything was persisted only after the
    broadcast. Now the intent is already durable, so the reconciler has something to
    resolve instead of the round being stuck with zero trace of what was attempted."""
    await _seed_paying_out_round(payout_session_factory)
    entries = [{"tx_hash": "44" * 32, "tx_pos": 0, "height": 10, "value": _POOL_AMOUNT_SATS + 100_000}]
    client = FakePayoutClient(entries, fail_broadcast=True)
    scheduler = RoundScheduler(payout_session_factory, FakePayoutListener(client))

    await scheduler._trigger_payout(1)

    assert not client.broadcasted
    async with payout_session_factory() as session:
        round_ = await session.get(Round, 1)
        assert round_.payout_txid is not None  # durable, even though the broadcast failed

        pending = (await session.scalars(select(PendingTransaction))).one()
        assert pending.kind == "payout"
        assert pending.status == "building"  # not lost — the reconciler resolves this
        assert pending.current_txid == round_.payout_txid

        events = [e.event_type for e in (await session.scalars(select(AuditLog))).all()]
        assert "payout_failed" in events


async def test_trigger_payout_skips_when_already_in_flight(payout_session_factory):
    """A second call for a round that already has a non-terminal payout
    PendingTransaction must not build (and broadcast) another one — that would pay
    the winner twice."""
    winner_id = await _seed_paying_out_round(payout_session_factory)
    async with payout_session_factory() as session:
        round_ = await session.get(Round, 1)
        round_.payout_txid = "already-sent-txid"
        session.add(
            PendingTransaction(
                kind="payout",
                round_id=1,
                current_txid="already-sent-txid",
                fee_rate_sat_vb=1,
                raw_tx_hex="00",
                status="pending",
            )
        )
        await session.commit()

    entries = [{"tx_hash": "55" * 32, "tx_pos": 0, "height": 10, "value": _POOL_AMOUNT_SATS + 100_000}]
    client = FakePayoutClient(entries)
    scheduler = RoundScheduler(payout_session_factory, FakePayoutListener(client))

    await scheduler._trigger_payout(1)

    assert not client.broadcasted
    async with payout_session_factory() as session:
        assert (await session.scalars(select(PendingTransaction))).all()  # still just the one seeded
        rows = (await session.scalars(select(PendingTransaction))).all()
        assert len(rows) == 1
        assert rows[0].current_txid == "already-sent-txid"


async def test_reserved_payout_outpoints_excludes_utxos_claimed_by_a_stale_payout(payout_session_factory):
    """A payout still "building"/"pending" for some round — most plausibly a stale
    one the reconciler hasn't abandoned yet — must keep its inputs off the table for
    a fresh payout attempt, or the same pool coins could be spent twice."""
    from embit import script
    from embit.transaction import Transaction, TransactionInput, TransactionOutput

    reserved_txid = "66" * 32
    raw_tx = (
        Transaction(
            vin=[TransactionInput(bytes.fromhex(reserved_txid), 2)],
            vout=[TransactionOutput(1_000_000, script.Script.from_address(_WINNER_ADDRESS))],
        )
        .serialize()
        .hex()
    )
    async with payout_session_factory() as session:
        session.add(
            PendingTransaction(
                kind="payout",
                round_id=99,
                current_txid="stale-payout-txid",
                fee_rate_sat_vb=1,
                raw_tx_hex=raw_tx,
                status="building",
            )
        )
        await session.commit()

        reserved = await _reserved_payout_outpoints(session)

    assert reserved == {(reserved_txid, 2)}


# --- B-26: a "paying_out" round must retry its payout automatically ---------------


async def test_trigger_payout_logs_a_failure_when_not_connected(payout_session_factory):
    """Before B-26, this early return logged nothing beyond a log line — invisible
    in /admin and unusable as a signal for an automatic retry."""
    await _seed_paying_out_round(payout_session_factory)
    scheduler = RoundScheduler(payout_session_factory, FakePayoutListener(client=None))

    await scheduler._trigger_payout(1)

    async with payout_session_factory() as session:
        entries = (await session.scalars(select(AuditLog).where(AuditLog.event_type == "payout_failed"))).all()
        assert len(entries) == 1
        assert entries[0].payload_json.count("electrum client not connected") == 1


async def test_trigger_payout_logs_a_failure_when_fee_address_missing(payout_session_factory):
    winner_id = await _seed_paying_out_round(payout_session_factory)
    async with payout_session_factory() as session:
        config = (await session.scalars(select(RoundConfig))).one()
        config.fee_address = ""
        await session.commit()

    entries = [{"tx_hash": "77" * 32, "tx_pos": 0, "height": 10, "value": _POOL_AMOUNT_SATS + 100_000}]
    scheduler = RoundScheduler(payout_session_factory, FakePayoutListener(FakePayoutClient(entries)))

    await scheduler._trigger_payout(1)

    async with payout_session_factory() as session:
        entry = (await session.scalars(select(AuditLog).where(AuditLog.event_type == "payout_failed"))).one()
        assert "no fee_address configured" in entry.payload_json
        assert entry.user_id == winner_id


async def test_tick_retries_a_stuck_paying_out_round_with_no_recent_failure(payout_session_factory):
    """The scenario B-26 exists for: a round stuck in "paying_out" (a prior failure,
    or a process restart mid-payout) with no non-terminal PendingTransaction. A
    fresh tick must retry rather than leaving it wedged forever."""
    await _seed_paying_out_round(payout_session_factory)
    entries = [{"tx_hash": "88" * 32, "tx_pos": 0, "height": 10, "value": _POOL_AMOUNT_SATS + 100_000}]
    client = FakePayoutClient(entries)
    scheduler = RoundScheduler(payout_session_factory, FakePayoutListener(client))

    await scheduler._tick()

    assert client.broadcasted
    async with payout_session_factory() as session:
        round_ = await session.get(Round, 1)
        assert round_.payout_txid is not None
        pending = (await session.scalars(select(PendingTransaction))).one()
        assert pending.status == "pending"


async def test_tick_throttles_retry_after_a_recent_payout_failure(payout_session_factory):
    """A payout that just failed must not be retried on the very next tick, or a
    persistently-broken payout (e.g. no fee_address) would spam a retry — and a
    fresh payout_failed audit entry — every _TICK_INTERVAL_SECONDS."""
    await _seed_paying_out_round(payout_session_factory)
    async with payout_session_factory() as session:
        session.add(
            AuditLog(
                event_type="payout_failed",
                payload_json='{"round_id": 1, "reason": "insufficient pool UTXOs"}',
                round_id=1,
                created_at=datetime.now(timezone.utc),
            )
        )
        await session.commit()

    entries = [{"tx_hash": "99" * 32, "tx_pos": 0, "height": 10, "value": _POOL_AMOUNT_SATS + 100_000}]
    client = FakePayoutClient(entries)
    scheduler = RoundScheduler(payout_session_factory, FakePayoutListener(client))

    await scheduler._tick()

    assert not client.broadcasted
    async with payout_session_factory() as session:
        assert (await session.scalars(select(PendingTransaction))).all() == []


async def test_tick_retries_once_the_throttle_window_has_elapsed(payout_session_factory):
    await _seed_paying_out_round(payout_session_factory)
    async with payout_session_factory() as session:
        session.add(
            AuditLog(
                event_type="payout_failed",
                payload_json='{"round_id": 1, "reason": "insufficient pool UTXOs"}',
                round_id=1,
                created_at=datetime.now(timezone.utc) - timedelta(seconds=120),
            )
        )
        await session.commit()

    entries = [{"tx_hash": "aa" * 32, "tx_pos": 0, "height": 10, "value": _POOL_AMOUNT_SATS + 100_000}]
    client = FakePayoutClient(entries)
    scheduler = RoundScheduler(payout_session_factory, FakePayoutListener(client))

    await scheduler._tick()

    assert client.broadcasted
    async with payout_session_factory() as session:
        pending = (await session.scalars(select(PendingTransaction))).one()
        assert pending.status == "pending"


# --- B-73 (RND-14 … RND-20, RND-33, RND-34): the commit-reveal draw ------------------------

_TIP = 100


class DrawListener:
    """What the draw asks of the listener: a tip that tests move, and the quorum's answer for a
    height (`blocks`: height -> hash, or None for "no agreement"). Every question is
    recorded so a test can assert it was always about the *fixed* draw height."""

    client = object()

    def __init__(self, tip_height: int = _TIP, blocks: dict[int, str | None] | None = None):
        self.tip_height = tip_height
        self.tip_header_hex = "00"
        self.blocks = blocks if blocks is not None else {}
        self.asked: list[int] = []
        self.reported: dict[int, str | None] = {}

    async def agreed_block_hash(self, height: int) -> tuple[str | None, str | None]:
        self.asked.append(height)
        return self.blocks.get(height), self.reported.get(height, self.blocks.get(height))


def _bet_txid(i: int) -> str:
    return f"{(0xA0 + i * 7) % 256:02x}" * 32


async def _seed_closing_round(session_factory, *, n: int = 3, draw_confirmations: int = 3, txids=None) -> int:
    """A real round opened by `open_new_round_if_needed` (so it carries its real encrypted
    secret), closed, with `n` confirmed participants whose txids are NOT in broadcast order."""
    from app.rounds.service import open_new_round_if_needed

    async with session_factory() as session:
        session.add(RoundConfig(fee_address=_FEE_ADDRESS, draw_confirmations=draw_confirmations))
        await session.commit()
        round_ = await open_new_round_if_needed(session)
        await session.commit()
        round_.status = "closing"
        round_.closed_at = datetime.now(timezone.utc)
        txids = txids or [_bet_txid(i) for i in range(n)]
        for i in range(n):
            user = User(username=f"player{i}", password_hash="x", derivation_index=i, address=_address(i))
            session.add(user)
            await session.flush()
            session.add(
                RoundParticipant(
                    round_id=round_.id,
                    user_id=user.id,
                    bet_amount_sats=1_000_000_000 - 5_000 * i,
                    bet_txid=txids[i],
                    bet_vout=0,
                    # broadcast order is the *reverse* of txid order for some of them, so an
                    # implementation still sorting by broadcast_at would get a different list
                    broadcast_at=datetime(2026, 1, 1, tzinfo=timezone.utc) + timedelta(seconds=i),
                    status="confirmed",
                )
            )
        await session.commit()
        return round_.id


def _address(i: int) -> str:
    return derive_pool_address(1000 + i)  # any distinct, valid PLM addresses


async def _enter_drawing_only(scheduler) -> list[int]:
    """Run `_close_and_draw` up to (not including) the draw itself."""
    entered: list[int] = []

    async def record(round_id: int) -> None:
        entered.append(round_id)

    scheduler._complete_draw = record
    return entered


async def test_closing_with_participants_commits_the_draw_and_enters_drawing(session_factory):
    """B-73 (RND-14, RND-33, DAT-65, AUD-37): in one transaction the list is ordered by bet txid
    and numbered, its fingerprint and the draw height (tip + 1) are stored, and the round
    enters `drawing` — before any block of the draw exists."""
    round_id = await _seed_closing_round(session_factory, n=3)
    listener = DrawListener(tip_height=_TIP)
    scheduler = RoundScheduler(session_factory, listener)
    entered = await _enter_drawing_only(scheduler)

    await scheduler._close_and_draw(round_id)

    assert entered == [round_id]
    async with session_factory() as session:
        round_ = await session.get(Round, round_id)
        assert round_.status == "drawing"
        assert round_.draw_height == _TIP + 1
        rows = (
            await session.execute(
                select(RoundParticipant.draw_position, RoundParticipant.bet_txid, RoundParticipant.user_id)
                .where(RoundParticipant.round_id == round_id)
                .order_by(RoundParticipant.draw_position)
            )
        ).all()
        assert [r.draw_position for r in rows] == [0, 1, 2]
        assert [r.bet_txid for r in rows] == sorted(_bet_txid(i) for i in range(3))  # ascending txid
        users = {u.id: u.address for u in (await session.scalars(select(User))).all()}
        # an independent recomputation of the fingerprint (RND-14)
        import hashlib

        text = "".join(f"{r.bet_txid}:0:{users[r.user_id]}\n" for r in rows)
        assert round_.participants_hash == hashlib.sha256(text.encode()).hexdigest()
        assert round_.pool_amount_sats == sum(1_000_000_000 - 5_000 * i for i in range(3))
        assert round_.drawing_started_at is not None

        committed = (await session.scalars(select(AuditLog).where(AuditLog.event_type == "draw_committed"))).all()
        assert len(committed) == 1 and committed[0].round_id == round_id
        import json

        assert json.loads(committed[0].payload_json) == {
            "participants_hash": round_.participants_hash,
            "draw_height": _TIP + 1,
            "participants": 3,
            "server_seed_hash": round_.server_seed_hash,
        }


async def test_participants_with_equal_broadcast_times_are_still_ordered_by_txid(session_factory):
    """RND-14 / DAT-66: the order is a function of public data only. The former order by
    `broadcast_at` had no tie-break: bets stamped with the same instant could be drawn in
    either order, so the draw could not be reproduced."""
    txids = ["cc" * 32, "aa" * 32, "bb" * 32]
    round_id = await _seed_closing_round(session_factory, n=3, txids=txids)
    async with session_factory() as session:
        for p in (await session.scalars(select(RoundParticipant))).all():
            p.broadcast_at = datetime(2026, 1, 1, tzinfo=timezone.utc)  # all identical
        await session.commit()
    scheduler = RoundScheduler(session_factory, DrawListener())
    await _enter_drawing_only(scheduler)

    await scheduler._close_and_draw(round_id)

    async with session_factory() as session:
        ordered = (
            await session.scalars(
                select(RoundParticipant.bet_txid)
                .where(RoundParticipant.round_id == round_id)
                .order_by(RoundParticipant.draw_position)
            )
        ).all()
    assert ordered == sorted(txids)


async def test_the_draw_height_is_fixed_when_entering_drawing_and_never_moves(session_factory):
    """RND-33: tip at closing + 1, stored on the round — the chain advancing afterwards does
    not change it."""
    round_id = await _seed_closing_round(session_factory)
    listener = DrawListener(tip_height=250)
    scheduler = RoundScheduler(session_factory, listener)
    await _enter_drawing_only(scheduler)
    await scheduler._close_and_draw(round_id)
    listener.tip_height = 400
    async with session_factory() as session:
        assert (await session.get(Round, round_id)).draw_height == 251


async def test_an_unknown_tip_keeps_the_round_closing_and_writes_one_audit_row(session_factory):
    """RND-19 / AUD-25: a tip of 0 is "unknown", not a base for the draw height. The round
    stays `closing` and is retried at the next step; the audit row is written once."""
    round_id = await _seed_closing_round(session_factory)
    listener = DrawListener(tip_height=0)
    scheduler = RoundScheduler(session_factory, listener)
    entered = await _enter_drawing_only(scheduler)

    await scheduler._close_and_draw(round_id)
    await scheduler._close_and_draw(round_id)

    assert entered == []
    async with session_factory() as session:
        round_ = await session.get(Round, round_id)
        assert round_.status == "closing" and round_.draw_height is None
        rows = (await session.scalars(select(AuditLog).where(AuditLog.event_type == "draw_baseline_tip_unknown"))).all()
        assert len(rows) == 1 and rows[0].payload_json == "{}"
    listener.tip_height = 80  # the height becomes known: the next step enters drawing
    await scheduler._close_and_draw(round_id)
    async with session_factory() as session:
        assert (await session.get(Round, round_id)).draw_height == 81


async def test_a_round_without_participants_closes_and_reveals_its_secret(session_factory):
    """DAT-64: nothing was drawn, so the secret is revealed at the close."""
    round_id = await _seed_closing_round(session_factory, n=0)
    scheduler = RoundScheduler(session_factory, DrawListener())
    await scheduler._close_and_draw(round_id)
    async with session_factory() as session:
        round_ = await session.get(Round, round_id)
        assert round_.status == "closed"
        assert round_.server_seed is not None
        from app.rounds.draw import server_seed_commitment

        assert server_seed_commitment(round_.server_seed) == round_.server_seed_hash


async def test_a_round_opened_before_the_commitment_existed_gets_a_secret_when_it_enters_drawing(session_factory):
    """Upgrade path: a round with no `server_seed_enc` (opened before B-73) must not wedge — it
    is given a secret as it enters `drawing`, still before any block of the draw exists."""
    round_id = await _seed_closing_round(session_factory)
    async with session_factory() as session:
        round_ = await session.get(Round, round_id)
        round_.server_seed_enc = round_.server_seed_hash = None
        await session.commit()
    scheduler = RoundScheduler(session_factory, DrawListener())
    await _enter_drawing_only(scheduler)
    await scheduler._close_and_draw(round_id)
    async with session_factory() as session:
        round_ = await session.get(Round, round_id)
        assert round_.status == "drawing" and round_.server_seed_enc and round_.server_seed_hash
        assert round_.server_seed is None


def _fake_clock(monkeypatch, listener, on_sleep=None):
    """Drive `_wait_for_draw_block` without real time: `datetime.now` is a clock that only
    `asyncio.sleep` advances, and `on_sleep(now)` lets the test move the chain."""
    import app.rounds.scheduler as scheduler_module

    start = datetime.now(timezone.utc)

    class _Clock:
        now = start

    monkeypatch.setattr(
        scheduler_module, "datetime", type("_D", (), {"now": staticmethod(lambda tz=None: _Clock.now)})
    )

    async def fake_sleep(seconds: float) -> None:
        _Clock.now += timedelta(seconds=seconds)
        if on_sleep is not None:
            on_sleep(_Clock.now - start)

    monkeypatch.setattr(scheduler_module.asyncio, "sleep", fake_sleep)
    return start


async def test_the_draw_waits_for_the_confirmations_on_the_draw_block(session_factory, monkeypatch):
    """RND-15: draw_height 101 with 3 confirmations needs a chain at 103 or more. The quorum is
    not even asked before that — and then it is asked about 101, not about the tip."""
    listener = DrawListener(tip_height=101, blocks={101: "ab" * 32})
    scheduler = RoundScheduler(session_factory, listener)

    def on_sleep(elapsed):
        if elapsed >= timedelta(seconds=10):
            listener.tip_height = 102
        if elapsed >= timedelta(seconds=20):
            listener.tip_height = 103

    start = _fake_clock(monkeypatch, listener, on_sleep)
    block_hash = await scheduler._wait_for_draw_block(1, 101, 3, start)

    assert block_hash == "ab" * 32
    assert listener.asked == [101]  # only once the chain reached 103 — and about the fixed height


async def test_a_disagreement_asks_the_same_height_again_and_is_audited_once_per_hash(session_factory, monkeypatch):
    """RND-16 / AUD-26: no agreement means waiting and asking for the *same* height — never
    moving to another block. The audit row is written once per distinct hash the active server
    reports, not on every request."""
    listener = DrawListener(tip_height=105, blocks={101: None})
    listener.reported[101] = "ee" * 32
    scheduler = RoundScheduler(session_factory, listener)

    def on_sleep(elapsed):
        if elapsed >= timedelta(seconds=30):
            listener.blocks[101] = "cd" * 32  # the quorum finally agrees

    start = _fake_clock(monkeypatch, listener, on_sleep)
    block_hash = await scheduler._wait_for_draw_block(1, 101, 3, start)

    assert block_hash == "cd" * 32
    assert set(listener.asked) == {101} and len(listener.asked) > 2
    async with session_factory() as session:
        rows = (await session.scalars(select(AuditLog).where(AuditLog.event_type == "draw_header_corroboration_failed"))).all()
        import json

        assert [json.loads(r.payload_json) for r in rows] == [{"height": 101, "reported_hash": "ee" * 32}]

        # a different hash reported later is a new fact worth a row of its own
    listener.blocks[101] = None
    listener.reported[101] = "ff" * 32

    def on_sleep_again(elapsed):
        if elapsed >= timedelta(seconds=15):
            listener.blocks[101] = "cd" * 32

    _fake_clock(monkeypatch, listener, on_sleep_again)
    await scheduler._wait_for_draw_block(1, 101, 3, datetime.now(timezone.utc))
    async with session_factory() as session:
        rows = (await session.scalars(select(AuditLog).where(AuditLog.event_type == "draw_header_corroboration_failed"))).all()
        assert [json.loads(r.payload_json)["reported_hash"] for r in rows] == ["ee" * 32, "ff" * 32]


async def test_a_block_relayer_cannot_choose_the_block_that_seeds_the_draw(session_factory, monkeypatch):
    """B-73: the active server announces tip 104 with a block of its choosing (and withholds
    the real one) while the quorum has nothing yet for the fixed draw height 101: the draw
    must not take 104's hash, nor 102's — it waits for the quorum at 101, however long."""
    listener = DrawListener(tip_height=104, blocks={101: None, 102: "11" * 32, 104: "44" * 32})
    scheduler = RoundScheduler(session_factory, listener)

    def on_sleep(elapsed):
        if elapsed >= timedelta(seconds=60):
            listener.blocks[101] = "01" * 32

    start = _fake_clock(monkeypatch, listener, on_sleep)
    block_hash = await scheduler._wait_for_draw_block(1, 101, 3, start)

    assert block_hash == "01" * 32  # the hash at 101, whatever the others looked like
    assert set(listener.asked) == {101}


async def test_the_stall_clock_runs_only_while_the_chain_does_not_advance(session_factory, monkeypatch):
    """RND-20: with 3 confirmations a round waits ~3 blocks. A chain producing a block every
    120s must not look stalled (the old rule counted from the start and fired in almost every
    round)."""
    listener = DrawListener(tip_height=101, blocks={101: "ab" * 32})
    scheduler = RoundScheduler(session_factory, listener)

    def on_sleep(elapsed):
        listener.tip_height = 101 + int(elapsed.total_seconds() // 300)  # a block every 300s

    start = _fake_clock(monkeypatch, listener, on_sleep)
    await scheduler._wait_for_draw_block(1, 101, 3, start)  # needs tip 103: ~600s > 360s overall

    async with session_factory() as session:
        assert (await session.scalars(select(AuditLog).where(AuditLog.event_type == "draw_stalled"))).all() == []


async def test_a_chain_that_stops_advancing_writes_draw_stalled_and_repeats(session_factory, monkeypatch):
    """RND-20 / AUD-27: payload {draw_height, current_tip_height, elapsed_seconds}; one row per
    threshold for as long as the stall lasts."""
    import app.rounds.scheduler as scheduler_module

    listener = DrawListener(tip_height=101, blocks={101: "ab" * 32})
    scheduler = RoundScheduler(session_factory, listener)
    threshold = scheduler_module._DRAW_STALL_THRESHOLD_SECONDS

    def on_sleep(elapsed):
        if elapsed >= timedelta(seconds=2 * threshold + 30):
            listener.tip_height = 103

    start = _fake_clock(monkeypatch, listener, on_sleep)
    await scheduler._wait_for_draw_block(1, 101, 3, start)

    async with session_factory() as session:
        import json

        rows = (await session.scalars(select(AuditLog).where(AuditLog.event_type == "draw_stalled"))).all()
        payloads = [json.loads(r.payload_json) for r in rows]
    assert len(payloads) == 2
    assert all(p["draw_height"] == 101 and p["current_tip_height"] == 101 for p in payloads)
    assert payloads[0]["elapsed_seconds"] >= threshold


async def test_a_quorum_that_never_agrees_writes_draw_stalled(session_factory, monkeypatch):
    """RND-20: "oppure il quorum continua a non concordare" — the confirmations are there but
    the servers disagree for the whole threshold."""
    import app.rounds.scheduler as scheduler_module

    listener = DrawListener(tip_height=110, blocks={101: None})
    scheduler = RoundScheduler(session_factory, listener)
    threshold = scheduler_module._DRAW_STALL_THRESHOLD_SECONDS

    def on_sleep(elapsed):
        if elapsed >= timedelta(seconds=threshold + 20):
            listener.blocks[101] = "ab" * 32

    start = _fake_clock(monkeypatch, listener, on_sleep)
    await scheduler._wait_for_draw_block(1, 101, 3, start)
    async with session_factory() as session:
        assert len((await session.scalars(select(AuditLog).where(AuditLog.event_type == "draw_stalled"))).all()) == 1


async def test_close_and_draw_waits_when_a_bet_appears_after_the_tick_check(session_factory):  # B-53
    """_tick counts in-flight bets in a session of its own, so a "building" row that
    commits between that count and the participant snapshot used to be invisible to
    both: the round drew and paid out without the bet, while its sats still landed in
    the pool. _close_and_draw re-checks in the same session it snapshots from, and
    must leave the round in "closing" for the next tick rather than draw."""
    async with session_factory() as session:
        session.add(RoundConfig(fee_address=""))
        session.add(Round(status="closing", opened_at=datetime.now(timezone.utc)))
        await session.commit()
        round_ = (await session.scalars(select(Round))).one()
        session.add(
            RoundParticipant(
                round_id=round_.id,
                user_id=1,
                bet_amount_sats=1_000_000_000,
                bet_txid="ab" * 32,
                status="building",  # committed a moment after _tick counted zero
            )
        )
        await session.commit()
        round_id = round_.id

    scheduler = RoundScheduler(session_factory, FakeListener())
    await scheduler._close_and_draw(round_id)

    async with session_factory() as session:
        round_ = await session.get(Round, round_id)
        assert round_.status == "closing"  # not drawn, and not closed as participant-less
        assert round_.winner_user_id is None
        events = [e.event_type for e in (await session.scalars(select(AuditLog))).all()]
        assert "round_closed" not in events
        assert "winner_drawn" not in events


async def test_tick_ignores_a_config_duration_edited_mid_round(session_factory):  # B-61
    """Lowering round_duration_seconds from 600 to 30 while a round is 300s in used
    to close that round on the spot, because the deadline was recomputed live from
    the config on every tick. The edit applies to the *next* round."""
    async with session_factory() as session:
        session.add(RoundConfig(fee_address="", round_duration_seconds=30))  # just lowered
        session.add(
            Round(
                status="open",
                opened_at=datetime.now(timezone.utc) - timedelta(seconds=300),
                duration_seconds=600,  # what this round opened with
            )
        )
        await session.commit()

    scheduler = RoundScheduler(session_factory, FakeListener())
    await scheduler._tick()

    async with session_factory() as session:
        round_ = (await session.scalars(select(Round))).one()
        assert round_.status == "open"  # still 300s to go, by its own clock