Check confirmation/existence via scripthash history, not verbose replies (B-41)

poll_once and reconcile.py's existence check both called
blockchain.transaction.get(txid, verbose=True). Several Electrum server
implementations and versions reject the verbose flag outright
("verbose transactions are currently unsupported"), which would have
meant no confirmations and no reconciliation ever running against such
a server, read as a plain transport error. reconcile.py additionally
decided whether to abandon a transaction - releasing its funds - by
substring-matching the error text ("missing", "not found", ...), which
only works against ElectrumX's specific wording.

Both now ask blockchain.scripthash.get_history for the address that
owns every input of the transaction (a user's own address for a
bet/withdrawal, the pool address for a payout) and look for the txid in
the result: present with height > 0 means confirmed, present with
height <= 0 means still in the mempool, absent means the server
doesn't know it. get_history is a plain, universally-supported Electrum
method, and "not in the list" replaces the old substring-matching
entirely - no more guessing at error wording to decide whether to
release funds. History is cached per scripthash within one pass, since
every "payout" row shares the same pool address.

New app/tx/pending_address.py factors out own_address_for (the
address derivation was previously duplicated informally inside
tx/broadcast.py's signing context) so confirmation.py and reconcile.py
share one definition instead of two that could compute different
addresses for the same row.

tests/unit/test_confirmation.py and test_reconcile.py needed real User
rows and a master-key bootstrap they didn't have before, since address
derivation is now exercised for real rather than assumed. Suite grows
from 217 to 222 tests. BUGS.md moves B-41 to Previously fixed - no
Medium-severity finding remains open.

tests/unit/test_reconcile.py

@@ -1,5 +1,10 @@
 """Regression tests for B-04 (and the "building" half of B-08): a transaction that
-never made it onto the chain must give the coins back instead of freezing them."""
+never made it onto the chain must give the coins back instead of freezing them.
+
+Also covers B-41: existence/reconciliation checks go through
+blockchain.scripthash.get_history rather than a verbose blockchain.transaction.get
+reply, so the fake clients below implement get_history.
+"""
 
 import pytest
 from embit import script
@@ -7,37 +12,60 @@ from embit.transaction import Transaction, TransactionInput, TransactionOutput
 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 app.db.models import AuditLog, PendingTransaction, RoundParticipant, User, UtxoEvent, Withdrawal
 from app.tx.reconcile import reconcile_once
 
 
 class UnknownTxClient:
-    """A server that doesn't know any of the txids it's asked about."""
+    """A server whose history for any address never includes our txid."""
 
-    async def get_transaction(self, txid: str, verbose: bool = False):
-        raise RuntimeError(f"missing transaction {txid}")
+    async def get_history(self, scripthash: str) -> list[dict]:
+        return []
 
 
 class KnownTxClient:
-    async def get_transaction(self, txid: str, verbose: bool = False):
-        return {"txid": txid, "confirmations": 0}
+    """A server whose history for the address includes our txid — mined or
+    still in the mempool doesn't matter for existence, only for confirmation
+    (which is tx/confirmation.py's concern, not reconcile.py's)."""
+
+    def __init__(self, txid: str = "betxid"):
+        self._txid = txid
+
+    async def get_history(self, scripthash: str) -> list[dict]:
+        return [{"tx_hash": self._txid, "height": 100}]
 
 
 class BrokenClient:
     """A transport failure — says nothing about whether the tx exists."""
 
-    async def get_transaction(self, txid: str, verbose: bool = False):
+    async def get_history(self, scripthash: str) -> list[dict]:
         raise ConnectionResetError("connection reset")
 
 
 @pytest.fixture
-async def session_factory():
+async def session_factory(tmp_path, monkeypatch):
+    # own_address_for (B-41) derives each row's address via the HD wallet rather
+    # than trusting the DB's address column, so reconcile_once now needs a real
+    # master key set up — same bootstrap test_broadcast.py uses.
+    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
 
 
 # A real (unsigned) transaction spending one input, built rather than hand-written
@@ -66,11 +94,19 @@ async def _seed_bet(
     participant_status: str,
     age_seconds: int,
     last_broadcast_age_seconds: int | None = None,
+    derivation_index: int = 0,
 ):
     from datetime import datetime, timedelta, timezone
 
+    from app.wallet.hd import derive_user_address
+
     async with session_factory() as session:
-        user = User(username="u", password_hash="x", derivation_index=0, address="plm1qtest")
+        user = User(
+            username="u",
+            password_hash="x",
+            derivation_index=derivation_index,
+            address=derive_user_address(derivation_index),
+        )
         session.add(user)
         await session.flush()
         session.add(
@@ -145,7 +181,7 @@ async def test_promotes_a_building_row_whose_tx_did_reach_the_chain(session_fact
         session_factory, pending_status="building", participant_status="building", age_seconds=300
     )
 
-    resolved = await reconcile_once(session_factory, KnownTxClient())
+    resolved = await reconcile_once(session_factory, KnownTxClient("betxid"))
     assert resolved == 1
 
     async with session_factory() as session:
@@ -219,8 +255,10 @@ async def test_abandoned_withdrawal_is_marked_failed_and_kept(session_factory):
     they can see it didn't go through."""
     from datetime import datetime, timedelta, timezone
 
+    from app.wallet.hd import derive_user_address
+
     async with session_factory() as session:
-        user = User(username="w", password_hash="x", derivation_index=1, address="plm1qtest2")
+        user = User(username="w", password_hash="x", derivation_index=1, address=derive_user_address(1))
         session.add(user)
         await session.flush()
         session.add(
@@ -264,3 +302,42 @@ async def test_abandoned_withdrawal_is_marked_failed_and_kept(session_factory):
         assert withdrawal.status == "failed"
         assert withdrawal.txid is None
         assert (await session.scalars(select(UtxoEvent))).one().spent_txid is None
+
+
+# --- B-41: existence checks now use get_history and share it across candidates
+# sharing the same address, instead of a per-tx verbose blockchain.transaction.get. --
+
+
+async def test_reconcile_once_caches_history_per_scripthash(session_factory):
+    """Two payout PendingTransaction rows always share the same pool address —
+    fetching its history twice in one pass would be wasteful and, at scale
+    across many candidates on one address, needlessly slow the whole tick."""
+    from datetime import datetime, timedelta, timezone
+
+    async with session_factory() as session:
+        old = datetime.now(timezone.utc) - timedelta(hours=7)
+        session.add(
+            PendingTransaction(
+                kind="payout", round_id=1, current_txid="payout-a", fee_rate_sat_vb=1,
+                raw_tx_hex=_RAW_TX, status="pending", broadcast_at=old, last_broadcast_at=old,
+            )
+        )
+        session.add(
+            PendingTransaction(
+                kind="payout", round_id=2, current_txid="payout-b", fee_rate_sat_vb=1,
+                raw_tx_hex=_RAW_TX, status="pending", broadcast_at=old, last_broadcast_at=old,
+            )
+        )
+        await session.commit()
+
+    call_count = {"n": 0}
+
+    class CountingClient:
+        async def get_history(self, scripthash: str) -> list[dict]:
+            call_count["n"] += 1
+            return [{"tx_hash": "payout-a", "height": 100}, {"tx_hash": "payout-b", "height": 100}]
+
+    resolved = await reconcile_once(session_factory, CountingClient())
+
+    assert resolved == 0  # both exist — nothing to abandon or promote (already "pending")
+    assert call_count["n"] == 1  # one call covered both rows sharing the pool address