Never swap the tip's hash sideways, and never split it from its height (B-64)

_apply_header's linkage check only fires on a single-block advance, so a header
at the height we already held one for was applied on nothing but its own
self-consistency — and that check, as header_meets_its_own_target's own docstring
says, a server can satisfy with a self-declared easy target. So the one value the
draw is seeded from could be replaced under us at the current height, by a reorg
at the tip or by a single server disagreeing with the rest, with no check able to
speak to it. Separately, a header carrying no hex set tip_header_hex back to None
while advancing tip_height, leaving the two describing different blocks — the
exact pairing that function exists to keep.

Both are now refused without ending the session, unlike the fabrication cases
above them: neither is evidence of a hostile server, and rotating away would cost
us the one connection that also credits deposits and broadcasts transactions.

- A same-height header is ignored (logged when it actually differs). The hash
  committed to for a height is not swapped under us; if ours turns out to be the
  orphan, corroborate_header already refuses to seed a draw from it and the draw
  waits for a further block.
- A hex-less header is ignored outright: nothing to validate, nothing to draw
  from. A server that only ever pushed heights now freezes the draw — visibly,
  via B-36's draw_stalled — instead of costing us the connection.

Because ignoring is not fatal, _run_once additionally refuses to publish the
client when the initial header leaves the tip still unknown, so this cannot
reopen B-63's window from the other side.

The two _run_once tests are bounded with asyncio.wait_for: without their guard
that call waits on session tasks nothing ends, and a regression must fail rather
than hang the suite.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

tests/unit/test_electrum_listener.py

@@ -247,6 +247,52 @@ def test_apply_header_rejects_one_that_does_not_chain_from_the_tip(session_facto
     assert (listener.tip_height, listener.tip_header_hex) == (100, header_100)  # untouched
 
 
+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)
+    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):
+        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
+
+
+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)
+    listener._apply_header({"height": 100, "hex": header_100})
+
+    listener._apply_header({"height": 100, "hex": header_100})  # must not raise
+
+    assert (listener.tip_height, listener.tip_header_hex) == (100, header_100)
+
+
+def test_apply_header_ignores_one_carrying_no_hex(session_factory, caplog):  # B-64
+    """A hex-less header can be neither validated nor drawn from, and applying its
+    height alone used to *clear* the hex we already had — leaving tip_height and
+    tip_header_hex describing different blocks, which is the one thing this function
+    exists to prevent."""
+    listener = ElectrumListener(lambda endpoint: None, session_factory, _ENDPOINTS)
+    header_100 = _mine_header("00" * 32)
+    listener._apply_header({"height": 100, "hex": header_100})
+
+    with caplog.at_level(logging.WARNING):
+        listener._apply_header({"height": 101})  # height only, no hex
+
+    assert (listener.tip_height, listener.tip_header_hex) == (100, header_100)  # pair intact
+    assert "no header hex" in caplog.text
+
+
 def test_apply_header_skips_linkage_check_across_a_height_gap(session_factory):
     """A reconnect (or the very first header of a session) hands us whatever the
     server's current tip is — which is legitimately not a single-block advance
@@ -634,6 +680,24 @@ async def test_run_once_publishes_the_client_only_once_the_tip_is_known(session_
         await run_once_task
 
 
+async def test_run_once_refuses_a_session_whose_initial_header_carries_no_hex(session_factory):  # B-64
+    """A hex-less header is ignored rather than fatal (a server that only pushes
+    heights must not cost us the connection that also credits deposits) — but on the
+    *first* header of a process there is no tip to fall back on, and publishing the
+    client anyway would hand consumers a connection whose chain position is unknown,
+    which is exactly what B-63 closed."""
+    client = _FakeConnectClient({"height": 100})  # no "hex"
+    listener = ElectrumListener(lambda endpoint: client, session_factory, _ENDPOINTS)
+
+    # Bounded: without the guard _run_once goes on to wait on the session's tasks,
+    # which nothing in this test ever ends — a regression must fail, not hang.
+    with pytest.raises(HeaderValidationError):
+        await asyncio.wait_for(listener._run_once(_ENDPOINTS[0]), timeout=5)
+
+    assert listener.client is None
+    assert (listener.tip_height, listener.tip_header_hex) == (0, None)
+
+
 async def test_run_once_leaves_no_client_published_when_the_first_header_is_rejected(
     session_factory,
 ):  # B-63
@@ -645,7 +709,7 @@ async def test_run_once_leaves_no_client_published_when_the_first_header_is_reje
     listener = ElectrumListener(lambda endpoint: client, session_factory, _ENDPOINTS)
 
     with pytest.raises(HeaderValidationError):
-        await listener._run_once(_ENDPOINTS[0])
+        await asyncio.wait_for(listener._run_once(_ENDPOINTS[0]), timeout=5)
 
     assert listener.client is None
     assert (listener.tip_height, listener.tip_header_hex) == (0, None)