A PIN-protected companion that hangs up on pairing is asked for its PIN

On the Cardputer Zero, connecting to an ESP32 companion that wants a PIN
stalled for a minute and never asked for one. Its link came up and its
services resolved; the UART subscribe made BlueZ ask MeshTerm's agent
for the PIN; with none given the agent refused, and the companion
disconnected -- it hangs up rather than refusing the subscribe. The
guard that holds back drops while bleak retries HCI 0x3e swallowed that
hang-up too, so meshcore waited out its timeout on a dead link.

The agent already records whether BlueZ asked it for the PIN. A drop
after that is now flagged, and the connect ends at once as a pairing
failure -- the words the PIN handling answers -- so the reader is asked
for the PIN in seconds. Drops with no pairing asked are held as before.

Also: a link that opens and then drops during service discovery
("failed to discover services, device disconnected", an HCI Connection
Timeout 1.7 s in, at -22 dBm as readily as at -86) is retried like a
link that never opened. It was surfacing on the first try, and the next
try usually connects.

Checked on the Cardputer against the companion: three connects with no
PIN each answer "requires a Bluetooth pairing PIN" in 2-7 s, where each
took the 60 s timeout before.

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

meshterm/core/bluez.py

@@ -205,7 +205,7 @@ def device_tail(address: str) -> str:
 
 
 @asynccontextmanager
-async def answering(address: str, pin: str | None) -> AsyncIterator[None]:
+async def answering(address: str, pin: str | None) -> AsyncIterator[PinAgent]:
     """Be BlueZ's default pairing agent for one device while a connect runs.
 
     BlueZ does not wait to be asked to pair. When the companion refuses the UART subscribe
@@ -230,7 +230,9 @@ async def answering(address: str, pin: str | None) -> AsyncIterator[None]:
         pin: Its PIN, or ``None`` to refuse.
 
     Yields:
-        Nothing; the agent is live for the body of the ``async with``.
+        The agent, live for the body of the ``async with``. Its ``asked`` says whether
+        BlueZ asked it for the PIN, which is how the connect tells a companion that hung
+        up on a pairing from a link that merely dropped.
     """
     bus = None
     agent = PinAgent(f"/net/meshterm/connect{os.getpid()}", pin, device_tail(address))
@@ -250,7 +252,7 @@ async def answering(address: str, pin: str | None) -> AsyncIterator[None]:
     except Exception as exc:  # noqa: BLE001 - no bus, no BlueZ: connect without an agent
         _log.debug("couldn't stand in as the BlueZ agent: %s", exc)
     try:
-        yield
+        yield agent
     finally:
         if bus is not None:
             try:

meshterm/core/connection.py

@@ -22,12 +22,12 @@ import re
 import socket
 import sys
 from abc import ABC, abstractmethod
-from collections.abc import AsyncIterator, Callable, Iterable
-from contextlib import AbstractAsyncContextManager, asynccontextmanager, nullcontext
+from collections.abc import AsyncIterator, Awaitable, Callable, Iterable
+from contextlib import AbstractAsyncContextManager, asynccontextmanager, nullcontext, suppress
 from dataclasses import dataclass, replace
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from . import transmit_gate
 from .channels import CHANNEL_SLOT_PROBE_CAP
@@ -199,7 +199,9 @@ _DISCARD_TIMEOUT_S = 2.0
 #: connection on Windows is intermittently flaky (a slow-advertising peripheral is missed by
 #: bleak's internal lookup, or the link-layer connect races the just-finished discovery scan);
 #: a single retry recovers the common case without meaningfully delaying a genuinely absent
-#: device. This retries only the *link open* — never a rejected PIN and never a mesh transmit.
+#: device. A link that opens and then drops during service discovery, which a Cardputer
+#: Zero's radio does now and then, is retried the same way. This retries only the *link
+#: open* — never a rejected PIN and never a mesh transmit.
 _BLE_CONNECT_ATTEMPTS = 2
 
 #: Pause between BLE link-open attempts (seconds), giving the OS radio a beat to settle.
@@ -725,10 +727,35 @@ _PAIRING_REFUSED = frozenset(
 )
 
 
+#: What a link that opened and then dropped before the session was up reads as. BlueZ's
+#: bleak backend says "failed to discover services, device disconnected" when the peer is
+#: lost during service discovery — on a Cardputer Zero an HCI Connection Timeout (0x08)
+#: 1.7 s in, at any signal strength, with no security exchange begun; the next attempt
+#: connected. A radio hiccup, so it is retried like a link that never opened.
+_BLE_DROP_HINTS = ("device disconnected",)
+
+
+def _chain_says(exc: BaseException, hints: tuple[str, ...]) -> bool:
+    """Whether ``exc``, or anything it was raised from, mentions one of ``hints``.
+
+    Walks the whole ``__cause__``/``__context__`` chain, so a bleak error wrapped by the
+    meshcore transport is still recognized.
+    """
+    seen: set[int] = set()
+    current: BaseException | None = exc
+    while current is not None and id(current) not in seen:
+        seen.add(id(current))
+        text = str(current).lower()
+        if any(hint in text for hint in hints):
+            return True
+        current = current.__cause__ or current.__context__
+    return False
+
+
 def _is_ble_auth_error(exc: BaseException) -> bool:
     """Return whether ``exc`` (or any it was raised from) is a BLE authentication rejection.
 
-    Walks the whole ``__cause__``/``__context__`` chain matching :data:`_BLE_AUTH_HINTS`, so a
+    Matches :data:`_BLE_AUTH_HINTS` along the exception chain, so a
     ``BleakGATTProtocolError`` wrapped by the meshcore transport is still recognized.
 
     Args:
@@ -737,15 +764,16 @@ def _is_ble_auth_error(exc: BaseException) -> bool:
     Returns:
         ``True`` if the failure is a missing/rejected pairing rather than a dropped link.
     """
-    seen: set[int] = set()
-    current: BaseException | None = exc
-    while current is not None and id(current) not in seen:
-        seen.add(id(current))
-        text = str(current).lower()
-        if any(hint in text for hint in _BLE_AUTH_HINTS):
-            return True
-        current = current.__cause__ or current.__context__
-    return False
+    return _chain_says(exc, _BLE_AUTH_HINTS)
+
+
+def _is_ble_link_drop(exc: BaseException) -> bool:
+    """Whether opening the link failed because it dropped, not because security refused it.
+
+    An authentication refusal can end in a disconnect too, so anything that reads as one
+    is never a drop: a wrong or missing PIN must reach the PIN handling, never a retry.
+    """
+    return not _is_ble_auth_error(exc) and _chain_says(exc, _BLE_DROP_HINTS)
 
 
 #: ``errno`` values a serial open fails with, by what they mean for the reader.
@@ -907,12 +935,21 @@ class _ConnectingClients(list):
         connecting: Whether ``connect()`` is still running; while it is, a disconnect is
             recorded here and *not* passed on to meshcore (see
             :func:`_hold_disconnects_while_connecting`).
+        hung_up: Set when the link dropped after BlueZ had asked for the PIN: the companion
+            refusing the pairing, which no retry inside bleak will undo.
     """
 
     connecting = True
 
+    def __init__(self, clients: Iterable[object] = ()) -> None:
+        """Start from ``clients`` (none, normally) and no hang-up."""
+        super().__init__(clients)
+        self.hung_up = asyncio.Event()
 
-def _hold_disconnects_while_connecting(connection) -> _ConnectingClients:  # noqa: ANN001
+
+def _hold_disconnects_while_connecting(  # noqa: ANN001
+    connection, pin_asked: Callable[[], bool] = lambda: False
+) -> _ConnectingClients:
     """Keep meshcore from dropping its bleak client while that client is still connecting.
 
     On Linux, BlueZ answers a link attempt that failed to synchronise
@@ -930,8 +967,16 @@ def _hold_disconnects_while_connecting(connection) -> _ConnectingClients:  # noq
     the link really cannot be made, and once ``connect()`` returns, disconnects reach
     meshcore as before, because the client keeps calling this same wrapper.
 
+    The one drop that is not bleak's to retry is the companion hanging up on a pairing: a
+    PIN-protected companion (an ESP32 one, measured on a Cardputer Zero) answers the UART
+    subscribe by asking for the PIN, and on a refused or wrong one it disconnects rather
+    than refusing the subscribe. Held back, that left ``connect()`` waiting out its timeout
+    on a dead link and never asking for the PIN. A drop after BlueZ asked for the PIN is
+    therefore flagged on ``hung_up``, for the connect to give up on as a pairing failure.
+
     Args:
         connection: The ``meshcore.BLEConnection`` about to connect.
+        pin_asked: Whether BlueZ has asked this connect's agent for the PIN yet.
 
     Returns:
         The record of clients seen, whose ``connecting`` flag the caller clears once
@@ -945,6 +990,10 @@ def _hold_disconnects_while_connecting(connection) -> _ConnectingClients:  # noq
     def _handle_disconnect(client) -> None:  # noqa: ANN001 - a bleak client
         seen.append(client)
         if seen.connecting:
+            if pin_asked():
+                _log.debug("BLE companion hung up on the pairing it asked for")
+                seen.hung_up.set()
+                return
             _log.debug("BLE link attempt dropped while connecting; bleak retries it")
             return
         forward(client)
@@ -955,6 +1004,42 @@ def _hold_disconnects_while_connecting(connection) -> _ConnectingClients:  # noq
     return seen
 
 
+class _PairingHungUp(Exception):
+    """The companion disconnected after asking for the PIN: a refused or missing pairing.
+
+    Worded to read as an authentication failure (:data:`_BLE_AUTH_HINTS`), because it is
+    one, so the connect's PIN handling takes it from here rather than the link retry.
+    """
+
+
+async def _unless_hung_up(connect: Awaitable[Any], hung_up: asyncio.Event) -> Any:
+    """Await ``connect``, unless the companion hangs up on its pairing first.
+
+    Args:
+        connect: The meshcore ``connect()`` coroutine.
+        hung_up: Set by :func:`_hold_disconnects_while_connecting` on that hang-up.
+
+    Returns:
+        What ``connect`` returned.
+
+    Raises:
+        _PairingHungUp: If the hang-up came first; ``connect`` is cancelled.
+    """
+    task = asyncio.ensure_future(connect)
+    waiter = asyncio.ensure_future(hung_up.wait())
+    try:
+        await asyncio.wait({task, waiter}, return_when=asyncio.FIRST_COMPLETED)
+        if task.done():
+            return task.result()
+        raise _PairingHungUp("authentication failed: the companion hung up on the pairing")
+    finally:
+        waiter.cancel()
+        if not task.done():
+            task.cancel()
+            with suppress(BaseException):
+                await task
+
+
 class _WriteRefusals(list):
     """Every exception a ``BLEConnection`` write raised, oldest first."""
 
@@ -1895,6 +1980,9 @@ class MeshCoreDevice(Device):
         #: its own reference the moment the peripheral vanishes — see :meth:`_release_ble_client`
         #: for why nothing else can close it, and what that costs when it stays open.
         self._ble_client = None  # type: ignore[var-annotated]  # bleak.BleakClient
+        #: The BlueZ agent answering this device's pairing while a connect runs (Linux; see
+        #: :meth:`_ble_pairing_agent`), so the connect can tell a pairing refusal from a drop.
+        self._ble_agent: object | None = None
         # Serializes channel reads. The meshcore library's get_channel waits for "the next
         # CHANNEL_INFO event" with no correlation to the index it asked for, and the dispatcher
         # fans that event to *every* in-flight waiter — so two concurrent reads both resolve on
@@ -2011,9 +2099,17 @@ class MeshCoreDevice(Device):
         Returns:
             The connected ``MeshCore`` client, or ``None`` if the peripheral never answered.
         """
-        async with self._ble_pairing_agent():
-            await self._pair_ble(force=False)
-            return await self._open_ble(mesh_core, allow_repair=True)
+        async with self._ble_pairing_agent() as agent:
+            self._ble_agent = agent
+            try:
+                await self._pair_ble(force=False)
+                return await self._open_ble(mesh_core, allow_repair=True)
+            finally:
+                self._ble_agent = None
+
+    def _pin_was_asked(self) -> bool:
+        """Whether BlueZ asked this connect's agent for the PIN (see :class:`bluez.PinAgent`)."""
+        return bool(getattr(self._ble_agent, "asked", False))
 
     def _ble_pairing_agent(self) -> AbstractAsyncContextManager[object]:
         """Answer BlueZ's own pairing requests for this device while connecting (Linux only).
@@ -2130,9 +2226,11 @@ class MeshCoreDevice(Device):
         ``BLEDevice`` (when the picker's scan produced one) is passed through so the client
         connects to it directly instead of re-discovering the address.
 
-        Only ``ConnectionError`` is retried: a PIN/bond rejection or any other GATT failure
-        propagates unchanged on the first attempt so the auth handling in :meth:`_open_ble`
-        (and a genuine wrong-PIN) is never looped. Each attempt builds its own client through
+        A link that opened and then dropped before the session was up (see
+        :func:`_is_ble_link_drop`) is retried the same way: it is the radio, not the device
+        refusing. Nothing else is: a PIN/bond rejection or any other GATT failure propagates
+        unchanged on the first attempt so the auth handling in :meth:`_open_ble` (and a
+        genuine wrong-PIN) is never looped. Each attempt builds its own client through
         :meth:`_connect_owned_ble`, which closes it before letting any failure out — a
         retried attempt therefore starts from a released link, never a leaked one.
 
@@ -2148,13 +2246,15 @@ class MeshCoreDevice(Device):
                 because this is the one failure where the device was never reached at all,
                 and "didn't answer as a MeshCore device" sent people looking at the firmware.
         """
-        last_exc: ConnectionError | None = None
+        last_exc: Exception | None = None
         for attempt in range(_BLE_CONNECT_ATTEMPTS):
             if attempt:
                 await asyncio.sleep(_BLE_CONNECT_RETRY_DELAY_S)
             try:
                 return await self._connect_owned_ble(mesh_core)
-            except ConnectionError as exc:
+            except Exception as exc:  # noqa: BLE001 - only a link failure is kept and retried
+                if not (isinstance(exc, ConnectionError) or _is_ble_link_drop(exc)):
+                    raise
                 _log.debug(
                     "BLE link to %s failed to open (attempt %d/%d): %s",
                     self._address,
@@ -2163,14 +2263,14 @@ class MeshCoreDevice(Device):
                     exc,
                 )
                 last_exc = exc
-        assert last_exc is not None  # the loop always runs; only ConnectionError falls through
+        assert last_exc is not None  # the loop always runs; only a link failure falls through
         where = self._address or "the selected Bluetooth device"
         _log.warning("BLE link to %s never opened: %s", where, last_exc)
         raise DeviceCommandError(
             f"couldn't open a Bluetooth link to {where} ({_BLE_CONNECT_ATTEMPTS} tries) — it "
-            "wasn't found, or didn't accept the connection. It may be out of range, powered "
-            "off, or connected to a phone or another computer (a companion takes one "
-            "connection at a time)."
+            "wasn't found, didn't accept the connection, or dropped it while connecting. It "
+            "may be out of range, powered off, or connected to a phone or another computer (a "
+            "companion takes one connection at a time)."
         ) from last_exc
 
     async def _connect_owned_ble(self, mesh_core):  # type: ignore[no-untyped-def]
@@ -2228,7 +2328,7 @@ class MeshCoreDevice(Device):
         # removed by hand. Pairing is ours to run on every platform that has an API for it
         # (see :meth:`_pair_ble`).
         connection = BLEConnection(address=self._address, device=self._ble_device, pin=None)
-        seen = _hold_disconnects_while_connecting(connection)
+        seen = _hold_disconnects_while_connecting(connection, self._pin_was_asked)
         refused = _record_write_refusals(connection)
         mc = mesh_core(
             connection,
@@ -2236,7 +2336,7 @@ class MeshCoreDevice(Device):
             auto_reconnect=False,
         )
         try:
-            started = await mc.connect()
+            started = await _unless_hung_up(mc.connect(), seen.hung_up)
         except BaseException:
             self._ble_client = _held_client(connection, seen)
             await MeshCoreDevice._discard_meshcore(mc)

tests/test_ble_pairing.py

@@ -427,3 +427,105 @@ def test_every_link_open_attempt_failing_raises_the_last_error() -> None:
     assert isinstance(excinfo.value.__cause__, ConnectionError)
     assert len(_FakeMeshCore.built) == conn_mod._BLE_CONNECT_ATTEMPTS
     assert all(not c.cx.link_open for c in _FakeMeshCore.built)
+
+
+class _BleakError(Exception):
+    """Stand-in for ``bleak.exc.BleakError``, which carries what happened only as text."""
+
+
+def test_a_link_that_drops_while_connecting_is_retried() -> None:
+    """BlueZ losing the peer during service discovery is the radio, so it gets the retry.
+
+    Seen on a Cardputer Zero: an HCI Connection Timeout 1.7 s in, with no security
+    exchange begun, and the next attempt connected.
+    """
+    _FakeMeshCore.reset(_BleakError("failed to discover services, device disconnected"), "ok")
+    dev = _device()
+
+    client = asyncio.run(dev._create_ble_with_retry(_FakeMeshCore))
+
+    assert len(_FakeMeshCore.built) == 2
+    assert _FakeMeshCore.built[0].cx.link_open is False, "the dropped attempt leaked"
+    assert client is _FakeMeshCore.built[1]
+
+
+def test_an_auth_refusal_that_ends_in_a_disconnect_is_never_retried() -> None:
+    """A refused PIN can drop the link too; it goes to the PIN handling, not round again."""
+    _FakeMeshCore.reset(_AuthError("Insufficient Authentication: device disconnected"), "ok")
+    dev = _device()
+
+    with pytest.raises(_AuthError):
+        asyncio.run(dev._create_ble_with_retry(_FakeMeshCore))
+
+    assert len(_FakeMeshCore.built) == 1
+
+
+def test_any_other_failure_is_not_retried() -> None:
+    """Only a link that never opened or that dropped is the radio's; the rest surface at once."""
+    _FakeMeshCore.reset(_BleakError("Characteristic 6e400003 was not found"), "ok")
+    dev = _device()
+
+    with pytest.raises(_BleakError):
+        asyncio.run(dev._create_ble_with_retry(_FakeMeshCore))
+
+    assert len(_FakeMeshCore.built) == 1
+
+
+class _Connection:
+    """A ``BLEConnection`` stand-in: just the drop handler the guard wraps."""
+
+    def __init__(self) -> None:
+        self.forwarded: list[object] = []
+
+    def handle_disconnect(self, client: object) -> None:
+        self.forwarded.append(client)
+
+
+def test_a_hang_up_after_the_pin_was_asked_is_flagged_not_held() -> None:
+    """The retries bleak makes are held back; a companion refusing its pairing is not one."""
+    asked = False
+    connection = _Connection()
+    seen = conn_mod._hold_disconnects_while_connecting(connection, lambda: asked)
+
+    connection.handle_disconnect("first try")  # bleak's 0x3e retry: held, nothing flagged
+    assert not seen.hung_up.is_set()
+    asked = True
+    connection.handle_disconnect("after the PIN")
+    assert seen.hung_up.is_set()
+    assert connection.forwarded == []  # still held while connecting
+
+
+def test_a_hang_up_ends_the_connect_as_a_pairing_failure() -> None:
+    """The connect stops waiting on a dead link and says why, in the PIN handling's words."""
+
+    async def scenario() -> bool:
+        hung_up = asyncio.Event()
+        cancelled = asyncio.Event()
+
+        async def connect() -> str:
+            try:
+                await asyncio.sleep(60)
+            except asyncio.CancelledError:
+                cancelled.set()
+                raise
+            return "never"
+
+        asyncio.get_running_loop().call_later(0.05, hung_up.set)
+        with pytest.raises(conn_mod._PairingHungUp) as excinfo:
+            await conn_mod._unless_hung_up(connect(), hung_up)
+        assert conn_mod._is_ble_auth_error(excinfo.value)
+        return cancelled.is_set()
+
+    assert asyncio.run(scenario()), "the abandoned connect was left running"
+
+
+def test_a_connect_that_finishes_first_is_returned() -> None:
+    """No hang-up, no change: the connect's own answer comes back."""
+
+    async def scenario() -> str:
+        async def connect() -> str:
+            return "started"
+
+        return await conn_mod._unless_hung_up(connect(), asyncio.Event())
+
+    assert asyncio.run(scenario()) == "started"