Follow and record chain reorganisations, pausing on deep ones (B-85, ELC-09, ELC-22, ELC-23, ELC-24, AUD-38, DAT-69)

A reorganisation was neither recorded nor acted upon, and a competing header at the tip's
own height was ignored for as long as the tip stood, leaving the listener on an orphan.

- A header at the height we hold with a different hash is followed through the same
  fork-point search as any other reorg (ELC-09) instead of being ignored; it still has to
  bring valid work and a connected chain.
- ElectrumListener._record_reorg writes `chain_reorg_detected` (depth, fork height, old and
  new tip height and hash) before the window moves, so a failed write ends the session and
  the next server's announcement of the same fork is recorded then. Depth 1 is a
  transaction of its own; depth >= 2 also pauses the lottery with paused_reason 'reorg'
  and writes `lottery_paused` in the same transaction (begin_immediate first).
- Docs: ELC-09/22/23/24, AUD-03/11/38, DAT-30/69, OPS-30, RND-04, ADR-26/52 lose their
  B-85 "da implementare" lines.

Tests changed because the spec changed: test_apply_header_ignores_a_competing_header_at_
the_current_tip_height encoded the B-64 behaviour that ELC-09 now reverses; it is replaced
by test_a_competing_header_at_the_tip_height_is_a_reorg_to_follow and the depth tests.

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

tests/unit/test_electrum_listener.py

@@ -18,7 +18,7 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 
 from app.db.base import Base
-from app.db.models import User, UtxoEvent
+from app.db.models import AuditLog, RoundConfig, User, UtxoEvent
 from app.electrum.client import ElectrumEndpoint
 from app.electrum.listener import ElectrumListener
 from app.rounds.draw import HeaderValidationError, header_hex_to_block_hash, header_meets_its_own_target
@@ -247,26 +247,6 @@ async def test_apply_header_rejects_one_that_does_not_chain_from_the_tip(session
     assert (listener.tip_height, listener.tip_header_hex) == (100, header_100)  # untouched
 
 
-async def test_apply_header_ignores_a_competing_header_at_the_current_tip_height(session_factory, caplog):  # B-64
-    """The linkage check only fires on a single-block advance, so a header at the
-    height we already hold one for used to be applied on nothing but its own
-    self-consistency — replacing the very hash a draw may be about to be seeded
-    with. Whichever it is (a reorg at the tip, or a server disagreeing with the
-    rest), the hash committed to for a height is not swapped under us; if ours is
-    the orphan, corroborate_header refuses to draw from it anyway."""
-    listener = ElectrumListener(lambda endpoint: None, session_factory, _ENDPOINTS)
-    header_100 = _mine_header("00" * 32)
-    await listener._apply_header({"height": 100, "hex": header_100})
-
-    competing_100 = _mine_header("11" * 32)  # same height, well-formed, different block
-    assert competing_100 != header_100
-    with caplog.at_level(logging.WARNING):
-        await listener._apply_header({"height": 100, "hex": competing_100})
-
-    assert (listener.tip_height, listener.tip_header_hex) == (100, header_100)  # untouched
-    assert "competing header" in caplog.text
-
-
 async def test_apply_header_treats_the_same_header_re_announced_as_a_no_op(session_factory):  # B-64
     listener = ElectrumListener(lambda endpoint: None, session_factory, _ENDPOINTS)
     header_100 = _mine_header("00" * 32)
@@ -1115,3 +1095,162 @@ async def test_a_header_easier_than_the_pow_limit_ends_the_session(session_facto
     with pytest.raises(HeaderValidationError):
         await fresh._apply_header({"height": 100, "hex": chain[100]}, client)
     assert (listener.tip_height, fresh.tip_height) == (100, 0)
+
+
+# --- B-85 (ELC-09, ELC-24, AUD-38): reorganisations are recorded, deep ones pause -------
+
+
+async def _reorg_events(session_factory) -> list[dict]:
+    import json
+
+    async with session_factory() as session:
+        rows = (await session.scalars(select(AuditLog).where(AuditLog.event_type == "chain_reorg_detected"))).all()
+        return [json.loads(row.payload_json) for row in rows]
+
+
+async def _config(session_factory) -> RoundConfig:
+    from app.rounds.config import get_round_config
+
+    async with session_factory() as session:
+        return await get_round_config(session)
+
+
+def _fork_at(chain_a: dict[int, str], fork_height: int, length: int, salt: int = 1) -> dict[int, str]:
+    """The server's chain: ours up to `fork_height`, then `length` blocks of its own."""
+    branch = _mine_chain(fork_height + 1, length, header_hex_to_block_hash(chain_a[fork_height]), salt=salt)
+    return {**{h: v for h, v in chain_a.items() if h <= fork_height}, **branch}
+
+
+async def test_a_one_block_reorg_is_followed_and_only_audited(session_factory):
+    """B-85, ELC-24: the tip 100 turns orphan and the server announces 101 on a branch from
+    99 — depth 1. The listener follows it and writes `chain_reorg_detected`; nothing is
+    paused, since a one-block reorganisation is ordinary."""
+    chain_a = _mine_chain(91, 10)
+    listener, client = await _listener_at(session_factory, chain_a, 100)
+    chain_b = _fork_at(chain_a, 99, 2)
+    client.chain = chain_b
+
+    await listener._apply_header({"height": 101, "hex": chain_b[101]}, client)
+
+    assert (listener.tip_height, listener.tip_header_hex) == (101, chain_b[101])
+    assert await _reorg_events(session_factory) == [
+        {
+            "depth": 1,
+            "fork_height": 99,
+            "old_tip_height": 100,
+            "old_tip_hash": header_hex_to_block_hash(chain_a[100]),
+            "new_tip_height": 101,
+            "new_tip_hash": header_hex_to_block_hash(chain_b[101]),
+        }
+    ]
+    config = await _config(session_factory)
+    assert (config.paused, config.paused_reason) == (False, None)
+
+
+async def test_a_competing_header_at_the_tip_height_is_a_reorg_to_follow(session_factory):
+    """B-85, ELC-09: a header for the height we hold with a *different* hash used to be
+    ignored (B-64), leaving the tip on an orphan. It is now followed like any fork."""
+    chain_a = _mine_chain(91, 10)
+    listener, client = await _listener_at(session_factory, chain_a, 100)
+    chain_b = _fork_at(chain_a, 99, 1)
+    assert chain_b[100] != chain_a[100]
+    client.chain = chain_b
+
+    await listener._apply_header({"height": 100, "hex": chain_b[100]}, client)
+
+    assert (listener.tip_height, listener.tip_header_hex) == (100, chain_b[100])
+    [event] = await _reorg_events(session_factory)
+    assert (event["depth"], event["fork_height"]) == (1, 99)
+
+
+async def test_a_competing_header_without_work_or_chain_is_not_followed(session_factory):
+    """ELC-09 / ELC-23: the competing header has to prove itself — a forged one that does
+    not join the server's own chain ends the session and changes nothing."""
+    chain_a = _mine_chain(91, 10)
+    listener, client = await _listener_at(session_factory, chain_a, 100)
+    stranger = _mine_chain(100, 1, "ee" * 32, salt=5)[100]
+    client.chain = {**chain_a, 100: stranger}
+
+    with pytest.raises(HeaderValidationError):
+        await listener._apply_header({"height": 100, "hex": stranger}, client)
+    assert listener.tip_header_hex == chain_a[100]
+    assert await _reorg_events(session_factory) == []
+
+
+async def test_a_reorg_of_depth_two_pauses_the_lottery_with_reason_reorg(session_factory):
+    """B-85, ELC-24: two of our blocks replaced (fork at 98, tip 100 -> 101) pauses the
+    lottery with `paused_reason = 'reorg'`, and the `lottery_paused` row is written with
+    the event. The round in progress is untouched; only the next one waits for the
+    operator."""
+    chain_a = _mine_chain(91, 10)
+    listener, client = await _listener_at(session_factory, chain_a, 100)
+    chain_b = _fork_at(chain_a, 98, 3)  # 99', 100', 101'
+    client.chain = chain_b
+
+    await listener._apply_header({"height": 101, "hex": chain_b[101]}, client)
+
+    assert listener.tip_height == 101
+    [event] = await _reorg_events(session_factory)
+    assert (event["depth"], event["fork_height"]) == (2, 98)
+    config = await _config(session_factory)
+    assert (config.paused, config.paused_reason) == (True, "reorg")
+    async with session_factory() as session:
+        reasons = [
+            row.payload_json
+            for row in (await session.scalars(select(AuditLog).where(AuditLog.event_type == "lottery_paused"))).all()
+        ]
+    assert reasons == ['{"reason": "reorg"}']
+
+
+async def test_a_deeper_reorg_overrides_an_earlier_operator_pause_reason(session_factory):
+    """DAT-69: an automatic reason is written even over a pause that is already active."""
+    chain_a = _mine_chain(91, 10)
+    listener, client = await _listener_at(session_factory, chain_a, 100)
+    async with session_factory() as session:
+        from app.rounds.config import get_round_config
+
+        config = await get_round_config(session)
+        config.paused, config.paused_reason = True, "operator"
+        await session.commit()
+    client.chain = _fork_at(chain_a, 96, 5)
+
+    await listener._apply_header({"height": 101, "hex": client.chain[101]}, client)
+
+    config = await _config(session_factory)
+    assert (config.paused, config.paused_reason) == (True, "reorg")
+    [event] = await _reorg_events(session_factory)
+    assert event["depth"] == 4
+
+
+async def test_a_fork_beyond_the_window_writes_no_event_and_pauses_nothing(session_factory):
+    """AUD-38: a fork deeper than 10 headers is not followed and writes no event — the
+    session ends instead."""
+    chain_a = _mine_chain(91, 10)
+    listener, client = await _listener_at(session_factory, chain_a, 100)
+    other = _mine_chain(91, 11, "dd" * 32, salt=2)
+    client.chain = other
+    with pytest.raises(HeaderValidationError):
+        await listener._apply_header({"height": 101, "hex": other[101]}, client)
+    assert await _reorg_events(session_factory) == []
+    assert (await _config(session_factory)).paused is False
+
+
+async def test_the_window_does_not_move_when_the_reorg_cannot_be_recorded(session_factory):
+    """A reorg is never followed without a trace: if the audit write fails the session
+    ends, and the next server announces the fork against an unchanged window."""
+    chain_a = _mine_chain(91, 10)
+    listener, client = await _listener_at(session_factory, chain_a, 100)
+    client.chain = _fork_at(chain_a, 98, 3)
+
+    def broken_session_factory():
+        raise RuntimeError("database is locked")
+
+    listener._session_factory = broken_session_factory
+    with pytest.raises(RuntimeError):
+        await listener._apply_header({"height": 101, "hex": client.chain[101]}, client)
+    assert (listener.tip_height, listener.tip_header_hex) == (100, chain_a[100])
+
+    listener._session_factory = session_factory  # the database is back: same fork, now recorded
+    await listener._apply_header({"height": 101, "hex": client.chain[101]}, client)
+    assert listener.tip_height == 101
+    assert len(await _reorg_events(session_factory)) == 1