fix(vehicle): make BLE pairing survive lost whitelist replies

AGENTS.md

@@ -133,6 +133,7 @@ Source protobuf definitions live in `proto/`; generated Python files live in `te
 - **`expects_data` splits BLE reply-waiting: VCSEC actuations return on the terminal ack**: a VCSEC read replies with a bare ACK **then** a data frame, but a VCSEC actuation (RKE/closure/wake via `_sendVehicleSecurity`) replies with a **single bare ACK only** - `_send` cannot tell the two apart at transport level, so the caller declares it. `_sendVehicleSecurity` passes `expects_data=False` down through `_command` into `_send` (`commands.py`/`bluetooth.py`); everything else (VCSEC reads via `_getVehicleSecurity`, all infotainment via `_send/_getInfotainment`, `_handshake`, `pair`) keeps the default `expects_data=True`. With `expects_data=False`, `_send` returns immediately on the matching ACK instead of waiting out `_ack_followup_timeout` (was a flat ~2s tax on every VCSEC actuation), and on a lost ack it raises `BluetoothTimeout` after the shorter `_actuation_timeout` (2s) rather than `_default_timeout` (5s) - the verify-by-state contract makes a longer wait pointless. Infotainment actuations never paid the tax (their `actionStatus` rides in `protobuf_message_as_bytes`, so `_send` returns on that data frame). The WAIT/fault retry threads `expects_data` through unchanged, so the double-execute exposure noted above is unaffected.
 - **`navigation_gps_request`'s `order` param is a raw int, not a callable enum**: `commands.py` used to build it as `NavigationGpsRequest.RemoteNavTripOrder(order)`, treating the protobuf nested-enum wrapper (`EnumTypeWrapper`) as if it were a callable Python `enum.IntEnum` class - it isn't, so every call raised `TypeError` before any message was sent (found live during PR-8; this method had never been exercised over BLE before). Fixed to pass `order=order` directly (matching the working sibling `navigation_gps_destination_request`), which protobuf accepts as a bare int for an enum field at runtime.
 - **`ReassemblingBuffer` resets on a >1s inter-chunk gap, not just on decode failure**: `bluetooth.py`'s `ReassemblingBuffer.receive_data` discards any in-progress partial frame if the next chunk arrives more than `STALE_CHUNK_TIMEOUT` (1s) after the previous one, mirroring Tesla's official Go SDK (`teslamotors/vehicle-command`, `pkg/connector/ble/ble.go`'s `rxTimeout`). Without this, a chunk dropped mid-message left a stale partial in the buffer that got prepended to the next message, corrupting it until a lucky decode failure resynced. This is a frame-integrity hardening, not a fix for the separate ack-loss behavior documented above (that's a stalled/silent link, which no buffer-side reset can recover).
+- **`pair()` confirms whitelisting two ways: one-shot reply OR verify-by-state poll**: the whitelist-op success is a single VCSEC frame, and it lands on a dead session (lost forever) if the BLE link cycles while the user walks to the car to approve - live-observed on HA/macOS CoreBluetooth, where `tesla_fleet_api` logged "Reconnecting to S..." every ~100s during a pending pair, and the plain one-shot wait hung despite car-side completion. `pair()` (`bluetooth.py`) keeps the reply as the fast path (waits one `poll_interval` for it) but, on a lost reply, polls `_pair_probe()` every `poll_interval` until an overall `timeout` (default 300s) elapses. The probe is a VCSEC `_handshake` with our own public key: it succeeds only once the key is whitelisted and faults `NotOnWhitelistFault` until then - `_pair_probe` maps any `TeslaFleetError` (incl. transport failures from a mid-wait reconnect) to "not yet", so polling survives reconnects. The whitelist op is written **exactly once** - never re-sent - because a re-send re-prompts the user (and the retry/double-execute hazard documented above applies). Deadline with neither path confirming raises a typed `BluetoothTimeout`. Default behavior, no new knob (`poll_interval` is a defaulted param).
 - **Cross-transport parity (cloud REST `VehicleFleet` vs BLE `Commands`)**: the same-named command on both paths should build a semantically equivalent instruction from identical args - a divergence there is a bug, but response *bodies* legitimately differ (REST JSON dict vs decoded protobuf) and are not. `tests/test_cross_transport_parity.py` locks the equivalence in with mocked-both-transports tests. Known **non-bug FORM differences** (do not "fix"): `set_scheduled_departure`'s `preconditioning_enabled`/`off_peak_charging_enabled` (no proto fields), `window_control` lat/lon and `navigation_sc_request` `id` (no proto fields), `navigation_request`'s `type`/`locale`/`timestamp_ms` (REST share-intent framing), and `media_volume_up` (no Tesla REST endpoint - BLE-only; cloud raises volume via `adjust_volume`). Two **open divergences left unfixed** pending live verification: `clear_pin_to_drive_admin` builds `DrivingClearSpeedLimitPinAction` (speed-limit PIN, not PIN-to-Drive - suspected mismapping, security-sensitive), and `navigation_gps_request`'s `order` is required on BLE but optional on cloud (signature mismatch; null-order wire semantics undecided).
 
 ## Maintaining this file

docs/bluetooth_vehicles.md

@@ -58,6 +58,13 @@ async def main():
 asyncio.run(main())
 ```
 
+`pair()` waits for the tap-to-approve on the vehicle: it returns as soon as the
+whitelist success reply arrives, but also polls whether the key became effective
+(every `poll_interval` seconds, default 5) up to an overall `timeout` (default
+300 seconds), so it still completes if that one-shot reply is lost to a BLE
+reconnect while you walk to the car. It raises `BluetoothTimeout` if neither
+confirms in time.
+
 ## Wake Up Vehicle
 
 You can wake up a `VehicleBluetooth` instance using the `wake_up` method. Here's a basic example to wake up a `VehicleBluetooth` instance:

tesla_fleet_api/tesla/vehicle/bluetooth.py

@@ -715,9 +715,25 @@ async def pair(
         self,
         role: Role = Role.ROLE_OWNER,
         form: KeyFormFactor = KeyFormFactor.KEY_FORM_FACTOR_CLOUD_KEY,
-        timeout: int = 60,
+        timeout: float = 300,
+        poll_interval: float = 5,
     ):
-        """Pair the key."""
+        """Pair the key, confirming completion by reply or by key-state polling.
+
+        The whitelist-operation success is a single VCSEC frame. If the BLE
+        session cycles while the user walks to the vehicle to approve, that
+        frame can land on a dead session and be lost, hanging a plain one-shot
+        wait. The reply stays the fast path when it survives, but completion is
+        also confirmed by polling whether our public key is now whitelisted - a
+        VCSEC handshake with our own key succeeds only once it is. The whitelist
+        op is written exactly once; it is never re-sent, which would re-prompt
+        the user, and the poll re-handshakes on a fresh connection so it
+        survives mid-wait reconnects. Raises ``BluetoothTimeout`` if neither
+        path confirms before ``timeout`` seconds elapse.
+        """
+
+        if poll_interval <= 0:
+            raise ValueError("poll_interval must be greater than 0")
 
         request = UnsignedMessage(
             WhitelistOperation=WhitelistOperation(
@@ -733,23 +749,95 @@ async def pair(
             protobuf_message_as_bytes=request.SerializeToString(),
             uuid=randbytes(16),
         )
-        resp = await self._send(msg, "protobuf_message_as_bytes", timeout=timeout)
+
+        deadline = time.monotonic() + timeout
+
+        # Fast path: write the op once and wait one interval for its one-shot
+        # reply. A lost reply falls through to polling rather than re-sending.
+        try:
+            resp = await self._send(
+                msg,
+                "protobuf_message_as_bytes",
+                timeout=max(0.0, min(poll_interval, deadline - time.monotonic())),
+            )
+            self._raise_for_whitelist_reply(resp)
+            return
+        except BluetoothTimeout:
+            pass
+
+        # Slow path: poll whether our key became effective on the vehicle.
+        while time.monotonic() < deadline:
+            if self._consume_late_whitelist_reply():
+                return
+            if await self._pair_probe():
+                return
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                break
+            await asyncio.sleep(min(poll_interval, remaining))
+
+        if self._consume_late_whitelist_reply():
+            return
+        if await self._pair_probe():
+            return
+
+        raise BluetoothTimeout
+
+    async def _pair_probe(self) -> bool:
+        """Return True once our public key is whitelisted on the vehicle.
+
+        A VCSEC handshake with our own public key completes only when that key
+        is on the vehicle's whitelist; until pairing is approved it faults with
+        ``NotOnWhitelistFault``. Any transport failure mid-poll (a dropped
+        session) is treated as 'not yet' so polling keeps running across
+        reconnects rather than aborting.
+        """
+        try:
+            return await self._handshake(Domain.DOMAIN_VEHICLE_SECURITY)
+        except TeslaFleetError:
+            return False
+
+    def _consume_late_whitelist_reply(self) -> bool:
+        queue = self._queues[Domain.DOMAIN_VEHICLE_SECURITY]
+        deferred: list[RoutableMessage] = []
+        try:
+            while True:
+                resp = queue.get_nowait()
+                try:
+                    respMsg = FromVCSECMessage.FromString(
+                        resp.protobuf_message_as_bytes
+                    )
+                except DecodeError:
+                    deferred.append(resp)
+                    continue
+                if respMsg.HasField("commandStatus") and respMsg.commandStatus.HasField(
+                    "whitelistOperationStatus"
+                ):
+                    self._raise_for_whitelist_reply(resp)
+                    return True
+                deferred.append(resp)
+        except asyncio.QueueEmpty:
+            return False
+        finally:
+            for resp in deferred:
+                queue.put_nowait(resp)
+
+    def _raise_for_whitelist_reply(self, resp: RoutableMessage) -> None:
+        """Raise the mapped fault if a whitelist-op reply reports one."""
         respMsg = FromVCSECMessage.FromString(resp.protobuf_message_as_bytes)
-        if respMsg.commandStatus.whitelistOperationStatus.whitelistOperationInformation:
-            if (
-                respMsg.commandStatus.whitelistOperationStatus.whitelistOperationInformation
-                < len(WHITELIST_OPERATION_STATUS)
-            ):
-                exception = WHITELIST_OPERATION_STATUS[
-                    respMsg.commandStatus.whitelistOperationStatus.whitelistOperationInformation
-                ]
-                if exception:
-                    raise exception
-            else:
-                raise WhitelistOperationStatus(
-                    f"Unknown whitelist operation failure: {respMsg.commandStatus.whitelistOperationStatus.whitelistOperationInformation}"
-                )
-        return
+        info = (
+            respMsg.commandStatus.whitelistOperationStatus.whitelistOperationInformation
+        )
+        if not info:
+            return
+        if info < len(WHITELIST_OPERATION_STATUS):
+            exception = WHITELIST_OPERATION_STATUS[info]
+            if exception:
+                raise exception
+        else:
+            raise WhitelistOperationStatus(
+                f"Unknown whitelist operation failure: {info}"
+            )
 
     async def wake_up(self):
         """Wake up the vehicle security computer.

tests/test_ble_pair.py

@@ -0,0 +1,174 @@
+"""Pairing over a mocked transport: reply fast-path vs. verify-by-state polling.
+
+``pair()`` confirms whitelisting either by the one-shot whitelist reply (fast
+path) or, when that frame is lost to a session cycle, by polling a VCSEC
+handshake with our own key (whitelisted keys handshake; unpaired keys fault).
+These tests drive both paths and the reconnect/deadline behaviour with the real
+``_send``/``_handshake`` seams mocked.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+from unittest.mock import AsyncMock, patch
+
+from tesla_fleet_api.exceptions import (
+    BluetoothTimeout,
+    BluetoothTransportError,
+    NotOnWhitelistFault,
+    WhitelistOperationWhitelistFull,
+)
+from tesla_fleet_api.tesla.vehicle.proto.universal_message_pb2 import (
+    Destination,
+    Domain,
+    RoutableMessage,
+)
+from tesla_fleet_api.tesla.vehicle.proto.vcsec_pb2 import (
+    CommandStatus,
+    FromVCSECMessage,
+    WHITELISTOPERATION_INFORMATION_NONE,
+    WHITELISTOPERATION_INFORMATION_WHITELIST_FULL,
+    WhitelistOperation_information_E,
+    WhitelistOperation_status,
+)
+
+from ble_mocked_transport import MockedBleTransportTestCase
+
+
+def whitelist_reply(
+    info: WhitelistOperation_information_E = WHITELISTOPERATION_INFORMATION_NONE,
+) -> RoutableMessage:
+    """A canned whitelist-op reply; ``info=0`` means success, else a fault code."""
+    body = FromVCSECMessage(
+        commandStatus=CommandStatus(
+            whitelistOperationStatus=WhitelistOperation_status(
+                whitelistOperationInformation=info
+            )
+        )
+    )
+    return RoutableMessage(
+        from_destination=Destination(domain=Domain.DOMAIN_VEHICLE_SECURITY),
+        protobuf_message_as_bytes=body.SerializeToString(),
+    )
+
+
+class PairTest(MockedBleTransportTestCase):
+    """Cover the whitelist reply fast-path and the verify-by-state poll loop."""
+
+    def _mock_handshake(self, vehicle: Any, *outcomes: Any) -> AsyncMock:
+        """Install a mocked ``_handshake`` scripting each poll's outcome."""
+        handshake = AsyncMock(side_effect=list(outcomes))
+        setattr(vehicle, "_handshake", handshake)
+        return handshake
+
+    async def test_reply_path_success(self) -> None:
+        """A surviving success reply returns without ever polling."""
+        vehicle, send = self.make_vehicle()
+        send.return_value = whitelist_reply()
+        handshake = self._mock_handshake(vehicle)
+
+        await vehicle.pair(poll_interval=0.01, timeout=1)
+
+        self.assertEqual(send.await_count, 1)
+        handshake.assert_not_awaited()
+
+    async def test_reply_path_fault_raises(self) -> None:
+        """A whitelist-op fault reply raises its mapped exception."""
+        vehicle, send = self.make_vehicle()
+        send.return_value = whitelist_reply(
+            info=WHITELISTOPERATION_INFORMATION_WHITELIST_FULL
+        )
+
+        with self.assertRaises(WhitelistOperationWhitelistFull):
+            await vehicle.pair(poll_interval=0.01, timeout=1)
+
+    async def test_reply_lost_poll_detects_success(self) -> None:
+        """A lost reply is recovered by the poll, without re-sending the op."""
+        vehicle, send = self.make_vehicle()
+        send.side_effect = BluetoothTimeout
+        handshake = self._mock_handshake(vehicle, NotOnWhitelistFault, True)
+
+        await vehicle.pair(poll_interval=0.01, timeout=1)
+
+        # Whitelist op written exactly once; re-sending would re-prompt the user.
+        self.assertEqual(send.await_count, 1)
+        self.assertEqual(handshake.await_count, 2)
+
+    async def test_late_reply_fault_raises_during_poll(self) -> None:
+        """A late whitelist-op fault raises its mapped exception while polling."""
+        vehicle, send = self.make_vehicle()
+
+        async def timeout_after_late_fault(*args: Any, **kwargs: Any) -> None:
+            getattr(vehicle, "_queues")[Domain.DOMAIN_VEHICLE_SECURITY].put_nowait(
+                whitelist_reply(info=WHITELISTOPERATION_INFORMATION_WHITELIST_FULL)
+            )
+            raise BluetoothTimeout
+
+        send.side_effect = timeout_after_late_fault
+        handshake = self._mock_handshake(vehicle, True)
+
+        with self.assertRaises(WhitelistOperationWhitelistFull):
+            await vehicle.pair(poll_interval=0.01, timeout=1)
+
+        self.assertEqual(send.await_count, 1)
+        handshake.assert_not_awaited()
+
+    async def test_late_reply_after_final_sleep_returns(self) -> None:
+        """A reply landing in the final sleep is drained before timing out."""
+        vehicle, send = self.make_vehicle()
+        send.side_effect = BluetoothTimeout
+        handshake = self._mock_handshake(vehicle, False)
+
+        original_sleep = asyncio.sleep
+
+        async def queue_reply_then_return(delay: float) -> None:
+            getattr(vehicle, "_queues")[Domain.DOMAIN_VEHICLE_SECURITY].put_nowait(
+                whitelist_reply()
+            )
+            await original_sleep(delay)
+
+        with patch(
+            "tesla_fleet_api.tesla.vehicle.bluetooth.asyncio.sleep",
+            new=queue_reply_then_return,
+        ):
+            await vehicle.pair(poll_interval=0.02, timeout=0.01)
+
+        self.assertEqual(send.await_count, 1)
+        self.assertEqual(handshake.await_count, 1)
+
+    async def test_nonpositive_poll_interval_raises(self) -> None:
+        """Invalid poll intervals are rejected before sending the whitelist op."""
+        vehicle, send = self.make_vehicle()
+
+        with self.assertRaisesRegex(ValueError, "poll_interval must be greater than 0"):
+            await vehicle.pair(poll_interval=0)
+
+        send.assert_not_awaited()
+
+    async def test_reconnect_midwait_poll_continues(self) -> None:
+        """A transport failure mid-poll is 'not yet', so polling keeps going."""
+        vehicle, send = self.make_vehicle()
+        send.side_effect = BluetoothTimeout
+        handshake = self._mock_handshake(
+            vehicle, BluetoothTransportError, NotOnWhitelistFault, True
+        )
+
+        await vehicle.pair(poll_interval=0.01, timeout=1)
+
+        self.assertEqual(send.await_count, 1)
+        self.assertEqual(handshake.await_count, 3)
+
+    async def test_deadline_failure(self) -> None:
+        """Never confirming before the deadline raises a typed BluetoothTimeout."""
+        vehicle, send = self.make_vehicle()
+        send.side_effect = BluetoothTimeout
+        handshake = self._mock_handshake(vehicle)
+        handshake.side_effect = NotOnWhitelistFault
+
+        with self.assertRaises(BluetoothTimeout):
+            await vehicle.pair(poll_interval=0.01, timeout=0.05)
+
+        # Still only one whitelist op the whole time.
+        self.assertEqual(send.await_count, 1)
+        self.assertGreater(handshake.await_count, 0)