Resume a round in drawing from the database after an exception or a restart (B-81, RND-01, RND-15, RND-36, ARC-27, OPS-09, OPS-13, LIM-01)

`_tick` skipped `drawing`: the wait for the draw block lived only inside the
`_close_and_draw` call that entered it, so any exception (a locked database writing an
audit row, a malformed header) or a restart left the round in `drawing` for ever with the
pool inside and no new round able to open.

Since B-73 everything the draw needs is committed on entering `drawing` (encrypted secret,
ordered list with `draw_position`, `participants_hash`, `draw_height`, `draw_confirmations`),
so `_tick` now calls `_complete_draw` for a round in `drawing`. `_complete_draw` rereads that
state, is idempotent (a round no longer `drawing`, or with a winner, is left alone; the
winner write is re-checked under BEGIN IMMEDIATE) and runs inside the single scheduler loop,
so the inline call from `_close_and_draw` and the tick can never run concurrently.

Tests (test_draw_proof.py): an exception in the quorum query, then a tick, completes with the
same height, list, fingerprint and the winner the public proof recomputes; a brand-new
scheduler (restart) resumes and a second tick draws and triggers nothing again; a failure in
the very transaction that stores the winner leaves no winner and no revealed secret, and the
next tick completes. Mutation-checked: without the `_tick` branch all three fail.

Docs: the B-81 `*Stato:*`/`*Oggi:*` lines are gone (RND-01, RND-15, RND-36, ARC-27, OPS-09,
ADR-12, ADR-25), OPS-13 no longer prescribes the manual `closing` workaround, LIM-01 is
struck, storico-bug.md and BUGS.md updated. No existing test changed.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

tests/unit/test_draw_proof.py

@@ -346,3 +346,178 @@ async def test_a_different_block_gives_a_different_draw_but_never_a_different_li
             winners.add(draw_index(draw_seed(seed, block, round_.participants_hash), 8))
     assert len(winners) > 1  # the block matters ...
     assert (await ac.get(f"/rounds/{round_id}/proof")).json()["participants_hash"] == list_hash  # ... the list does not move
+
+
+# ---- B-81 (RND-01, RND-15, RND-36, ARC-27, OPS-09): `_tick` resumes a round in `drawing` ----
+
+
+class _DrawListener(_Listener):
+    """A chain whose tip is already past the confirmations, so a resumed draw needs no waiting."""
+
+    tip_height = 503
+
+
+async def _draw_state(session_factory, round_id: int) -> dict:
+    async with session_factory() as session:
+        round_ = await session.get(Round, round_id)
+        rows = (
+            await session.scalars(
+                select(RoundParticipant)
+                .where(RoundParticipant.round_id == round_id)
+                .order_by(RoundParticipant.draw_position)
+            )
+        ).all()
+        return {
+            "status": round_.status,
+            "draw_height": round_.draw_height,
+            "participants_hash": round_.participants_hash,
+            "positions": [(p.user_id, p.draw_position, p.bet_txid) for p in rows],
+            "winner": round_.winner_user_id,
+            "seed_hash": round_.server_seed_hash,
+        }
+
+
+async def test_a_tick_resumes_a_drawing_round_after_an_exception_with_the_same_draw(client):
+    """B-81 (RND-36, ARC-27): an exception in the middle of the draw (here the quorum query
+    failing, as a locked database would) leaves the round in `drawing` with everything fixed;
+    the next tick — same scheduler, nothing remembered in memory — completes it with the same
+    height, the same list and the winner the public proof recomputes. Without the `drawing`
+    branch of `_tick` the round stays in `drawing` for ever."""
+    ac, session_factory = client
+    round_id = await _open_round_with_bets(session_factory, n=6)
+
+    class _Flaky(_DrawListener):
+        failing = True
+
+        async def agreed_block_hash(self, height):
+            if self.failing:
+                raise RuntimeError("database is locked")
+            return await super().agreed_block_hash(height)
+
+    listener = _Flaky({501: _BLOCK_HASH})
+    listener.tip_height = 500
+    scheduler = RoundScheduler(session_factory, listener)
+    scheduler._trigger_payout = _noop_payout
+
+    async def fast_sleep(_seconds):
+        listener.tip_height = 503
+
+    import app.rounds.scheduler as scheduler_module
+
+    original_sleep = scheduler_module.asyncio.sleep
+    scheduler_module.asyncio.sleep = fast_sleep
+    try:
+        try:
+            await scheduler._close_and_draw(round_id)
+            raise AssertionError("the injected failure should have escaped")
+        except RuntimeError:
+            pass
+        before = await _draw_state(session_factory, round_id)
+        assert before["status"] == "drawing" and before["winner"] is None
+
+        listener.failing = False
+        await scheduler._tick()  # the run() loop's next step
+    finally:
+        scheduler_module.asyncio.sleep = original_sleep
+
+    after = await _draw_state(session_factory, round_id)
+    assert after["status"] == "paying_out" and after["winner"] is not None
+    for fixed in ("draw_height", "participants_hash", "positions", "seed_hash"):
+        assert after[fixed] == before[fixed]
+    proof = (await ac.get(f"/rounds/{round_id}/proof")).json()
+    _, address = _independent_winner(proof)
+    assert proof["winner_address"] == address
+
+
+async def test_a_restarted_scheduler_resumes_a_drawing_round_and_a_second_tick_draws_nothing(client):
+    """B-81 (RND-36, OPS-09): a brand-new scheduler (the process restarted) picks the round up
+    from the database alone. Once it is `paying_out` a further tick neither draws again nor
+    starts a second payout trigger from the draw path, and the winner does not change."""
+    ac, session_factory = client
+    round_id = await _open_round_with_bets(session_factory, n=4)
+
+    entry_listener = _DrawListener({})
+    entry_listener.tip_height = 500
+    entry = RoundScheduler(session_factory, entry_listener)
+
+    async def stop(_round_id: int) -> None:
+        return None
+
+    entry._complete_draw = stop  # the process "dies" right after the transition
+    await entry._close_and_draw(round_id)
+    before = await _draw_state(session_factory, round_id)
+    assert before["status"] == "drawing"
+
+    listener = _DrawListener({501: _BLOCK_HASH})
+    restarted = RoundScheduler(session_factory, listener)
+    payouts: list[int] = []
+
+    async def record_payout(rid: int) -> None:
+        payouts.append(rid)
+
+    restarted._trigger_payout = record_payout
+    retries: list[int] = []
+
+    async def record_retry(rid: int) -> None:
+        retries.append(rid)
+
+    restarted._retry_payout_if_due = record_retry
+    await restarted._tick()
+    after_first = await _draw_state(session_factory, round_id)
+    assert after_first["status"] == "paying_out"
+    assert after_first["draw_height"] == before["draw_height"] == 501
+    assert after_first["participants_hash"] == before["participants_hash"]
+    assert after_first["positions"] == before["positions"]
+    assert payouts == [round_id]
+    asked_after_first = list(listener.asked)
+
+    await restarted._tick()  # paying_out now: handled by the payout retry, not by a new draw
+    assert listener.asked == asked_after_first
+    assert payouts == [round_id] and retries == [round_id]
+    assert (await _draw_state(session_factory, round_id))["winner"] == after_first["winner"]
+
+
+async def test_resuming_after_a_failure_inside_the_draw_transaction_leaves_no_half_state(client, monkeypatch):
+    """B-81 (RND-36): the exception comes from the very step that stores the winner. The
+    transaction rolls back as a whole — no winner, no revealed secret, still `drawing` — and
+    the next tick completes it."""
+    ac, session_factory = client
+    round_id = await _open_round_with_bets(session_factory, n=3)
+    listener = _DrawListener({501: _BLOCK_HASH})
+    scheduler = RoundScheduler(session_factory, listener)
+    scheduler._trigger_payout = _noop_payout
+
+    entry_listener = _DrawListener({})
+    entry_listener.tip_height = 500
+    entry = RoundScheduler(session_factory, entry_listener)
+
+    async def stop(_round_id: int) -> None:
+        return None
+
+    entry._complete_draw = stop
+    await entry._close_and_draw(round_id)
+
+    import app.rounds.scheduler as scheduler_module
+
+    real_audit = scheduler_module.write_audit_log
+
+    async def failing_audit(session, event_type, *args, **kwargs):
+        if event_type == "winner_drawn":
+            raise RuntimeError("database is locked")
+        return await real_audit(session, event_type, *args, **kwargs)
+
+    monkeypatch.setattr(scheduler_module, "write_audit_log", failing_audit)
+    try:
+        await scheduler._tick()
+        raise AssertionError("the injected failure should have escaped")
+    except RuntimeError:
+        pass
+    async with session_factory() as session:
+        round_ = await session.get(Round, round_id)
+        assert (round_.status, round_.winner_user_id, round_.server_seed) == ("drawing", None, None)
+
+    monkeypatch.setattr(scheduler_module, "write_audit_log", real_audit)
+    await scheduler._tick()
+    async with session_factory() as session:
+        round_ = await session.get(Round, round_id)
+        assert round_.status == "paying_out" and round_.winner_user_id is not None and round_.server_seed