feat(bluetooth): add idle BLE keepalive

AGENTS.md

@@ -134,6 +134,7 @@ Source protobuf definitions live in `proto/`; generated Python files live in `te
 - **`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).
+- **Idle BLE keepalive (`keepalive_interval`)**: an idle held BLE link to the vehicle drops at ~42s mean lifetime (link supervision timeout ~720ms underneath, bluetoothd-verified on macOS CoreBluetooth); a single trivial passive GATT read every ~20s extends lifetime ~10x (~400s observed). `VehicleBluetooth.__init__`'s `keepalive_interval` (default `DEFAULT_KEEPALIVE_INTERVAL` = 20.0, `None`/`0` disables; threaded through `Vehicles`/`VehiclesBluetooth.create*`) starts one asyncio task per connection (`_keepalive_loop`, `bluetooth.py`) that reads `VERSION_UUID` only after `keepalive_interval` seconds of genuine GATT idleness. **Idle-triggered, not periodic**: `_last_activity` is bumped on every `_send` write and every `_on_notify` frame, so an active session never gets extra traffic; the loop recomputes the wait each pass and fires only when idle. The read is **bounded** (`_keepalive_timeout`, 2s) and **best-effort** - a prior un-timed RSSI read hung indefinitely against a sleeping car, so every attempt carries a timeout and swallows all failures (`_keepalive_read` catches `Exception`, never `CancelledError`); a failed keepalive never raises into user code, never triggers reconnect (the existing `connect_if_needed` machinery owns recovery), and never wakes the car. Task lifecycle is tied to the connection: started at the end of `connect()` (after `start_notify`), cancelled-and-awaited in `disconnect()` and restarted cleanly on reconnect (`_start_keepalive`/`_stop_keepalive`). **Sleep tradeoff**: these reads keep an *awake* car awake and defer vehicle sleep - consumers wanting the car to sleep should disable keepalive or disconnect when idle. Tests in `tests/test_ble_keepalive.py`.
 - **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

README.md

@@ -166,6 +166,12 @@ asyncio.run(main())
 
 For more detailed examples, see [Bluetooth for Vehicles](docs/bluetooth_vehicles.md).
 
+`VehicleBluetooth` keeps a held BLE connection alive during idle periods by
+default with a passive GATT read about every 20 seconds. Pass
+`keepalive_interval=None` (or `0`) when creating the vehicle to disable it;
+leaving it enabled can keep an already-awake car awake longer, so disconnect or
+disable keepalive when vehicle sleep is preferred.
+
 BLE connect/notify and GATT write failures from `VehicleBluetooth` raise
 `BluetoothTransportError`, a `TeslaFleetError` subclass, with the original
 transport exception chained as `__cause__`. Catch `TeslaFleetError` to handle

docs/bluetooth_vehicles.md

@@ -39,6 +39,35 @@ async def main():
 asyncio.run(main())
 ```
 
+## Keeping the Connection Alive (`keepalive_interval`)
+
+An idle held BLE link to the vehicle drops on its own after roughly 42 seconds
+on average. To hold it open, `VehicleBluetooth` issues a minimal passive GATT
+read (of the version characteristic - never a command, never anything signed or
+mutating) after `keepalive_interval` seconds without traffic. A single such read
+every 20 seconds keeps the link alive about ten times longer.
+
+The keepalive is idle-triggered: any real send or received frame resets the
+timer, so an active session never gets extra traffic and the read fires only
+during genuine idleness. It is bounded and best-effort - a read that cannot
+complete (for example against a sleeping car) is swallowed and never raises into
+your code or forces a reconnect, and it never wakes the vehicle or masks a
+sleeping car. Link recovery stays owned by the normal reconnect path.
+
+`keepalive_interval` defaults to about 20 seconds and is accepted by the
+constructor and by `vehicles.create(...)` / `vehicles.createBluetooth(...)`.
+Pass `None` or `0` to disable it.
+
+```python
+vehicle = tesla_bluetooth.vehicles.create("<vin>", keepalive_interval=20.0)
+# or disable it:
+vehicle = tesla_bluetooth.vehicles.create("<vin>", keepalive_interval=None)
+```
+
+Tradeoff: because these reads generate link traffic, they keep an already-awake
+car awake and defer vehicle sleep. If you want the vehicle to sleep while idle,
+disable keepalive or disconnect when you have no work for it.
+
 ## Pair Vehicle
 
 You can pair a `VehicleBluetooth` instance using the `pair` method. Here's a basic example to pair a `VehicleBluetooth` instance:

tesla_fleet_api/tesla/vehicle/bluetooth.py

@@ -88,6 +88,10 @@
 NAME_UUID = "00002a00-0000-1000-8000-00805f9b34fb"
 APPEARANCE_UUID = "00002a01-0000-1000-8000-00805f9b34fb"
 
+# An idle held BLE link to the vehicle drops at ~42s mean; a trivial GATT read
+# every 20s keeps it alive ~10x longer. See AGENTS.md for the measured evidence.
+DEFAULT_KEEPALIVE_INTERVAL = 20.0
+
 if TYPE_CHECKING:
     from tesla_fleet_api.tesla.tesla import Tesla
 
@@ -294,10 +298,22 @@ class VehicleBluetooth(Commands[BluetoothParentT], Generic[BluetoothParentT]):
     cannot be derived or read - true toggles, relative steps, and ack-only
     actions - re-raise the timeout unchanged, exactly as with verification off
     (the default).
+
+    ``keepalive_interval`` (default ~20s, ``None``/``0`` disables) keeps an
+    otherwise idle held connection from dropping: after that many seconds with
+    no real GATT traffic, a minimal passive characteristic read is issued to
+    reset the link supervision timer. It is idle-triggered - any real send or
+    received frame resets the timer, so an active session pays nothing - and it
+    is bounded and best-effort: a failed read (e.g. against a sleeping car) is
+    swallowed and never raises into caller code or forces a reconnect; the
+    existing reconnect machinery owns link recovery. Note the tradeoff: these
+    reads keep an *awake* car awake and so defer vehicle sleep. Callers that
+    want the car to sleep should disable keepalive or disconnect when idle.
     """
 
     ble_name: str
     verify_commands: bool
+    keepalive_interval: float | None
     device: BLEDevice | None = None
     client: BleakClient | None = None
     _queues: dict[Domain, asyncio.Queue[RoutableMessage]]
@@ -309,6 +325,10 @@ class VehicleBluetooth(Commands[BluetoothParentT], Generic[BluetoothParentT]):
     # A lost actuation ack is inconclusive; the contract is verify-by-state, so
     # a shorter wait than a data read's before raising loses nothing.
     _actuation_timeout: float = 2
+    # Bounded so a keepalive read against a sleeping car can never hang the loop.
+    _keepalive_timeout: float = 2
+    _keepalive_task: asyncio.Task[None] | None = None
+    _last_activity: float = 0.0
 
     def __init__(
         self,
@@ -317,9 +337,11 @@ def __init__(
         key: ec.EllipticCurvePrivateKey | None = None,
         device: BLEDevice | None = None,
         verify_commands: bool = False,
+        keepalive_interval: float | None = DEFAULT_KEEPALIVE_INTERVAL,
     ):
         super().__init__(parent, vin, key)
         self.verify_commands = verify_commands
+        self.keepalive_interval = keepalive_interval
         self.ble_name = "S" + hashlib.sha1(vin.encode("utf-8")).hexdigest()[:16] + "C"
         self._queues = {
             Domain.DOMAIN_VEHICLE_SECURITY: asyncio.Queue(),
@@ -376,6 +398,7 @@ async def connect(self, max_attempts: int = MAX_CONNECT_ATTEMPTS) -> None:
                 services=[SERVICE_UUID],
             )
             await self.client.start_notify(READ_UUID, self._on_notify)
+            await self._start_keepalive()
         # bleak-esphome converts an aioesphomeapi transport timeout into a
         # builtin TimeoutError, not a BleakError, so catch both to keep every
         # connect transport failure within TeslaFleetError.
@@ -391,6 +414,7 @@ async def connect(self, max_attempts: int = MAX_CONNECT_ATTEMPTS) -> None:
 
     async def disconnect(self) -> bool:
         """Disconnect from the Tesla BLE device."""
+        await self._stop_keepalive()
         if not self.client:
             return False
         await self.client.disconnect()
@@ -403,6 +427,55 @@ async def connect_if_needed(self, max_attempts: int = MAX_CONNECT_ATTEMPTS) -> N
                 LOGGER.info(f"Reconnecting to {self.ble_name}")
                 await self.connect(max_attempts=max_attempts)
 
+    async def _start_keepalive(self) -> None:
+        """Start the idle keepalive task for this connection, if enabled."""
+        await self._stop_keepalive()
+        if not self.keepalive_interval:
+            return
+        self._last_activity = time.monotonic()
+        self._keepalive_task = asyncio.ensure_future(self._keepalive_loop())
+
+    async def _stop_keepalive(self) -> None:
+        """Cancel and await the keepalive task so it never outlives the link."""
+        task = self._keepalive_task
+        self._keepalive_task = None
+        if task is None:
+            return
+        task.cancel()
+        try:
+            await task
+        except (asyncio.CancelledError, Exception):
+            pass
+
+    async def _keepalive_loop(self) -> None:
+        """Read a trivial characteristic after each span of GATT idleness."""
+        assert self.keepalive_interval is not None
+        while True:
+            idle_for = time.monotonic() - self._last_activity
+            wait = self.keepalive_interval - idle_for
+            if wait > 0:
+                await asyncio.sleep(wait)
+                continue
+            await self._keepalive_read()
+            # The read itself is activity, so the next span starts from now.
+            self._last_activity = time.monotonic()
+
+    async def _keepalive_read(self) -> None:
+        """Issue one bounded passive read; swallow every failure.
+
+        A keepalive failure must never surface to callers or trigger reconnects
+        - link recovery is owned by ``connect_if_needed`` - and a sleeping car
+        must stay detectable, so a read that cannot complete is simply dropped.
+        """
+        client = self.client
+        if client is None or not client.is_connected:
+            return
+        try:
+            async with asyncio.timeout(self._keepalive_timeout):
+                await client.read_gatt_char(VERSION_UUID)
+        except Exception as e:
+            LOGGER.debug(f"Keepalive read failed: {e}")
+
     async def __aenter__(self) -> VehicleBluetooth[BluetoothParentT]:
         """Enter the async context."""
         await self.connect()
@@ -422,6 +495,7 @@ def _on_notify(self, sender: BleakGATTCharacteristic, data: bytearray) -> None:
         if sender.uuid != READ_UUID:
             LOGGER.error(f"Unexpected sender: {sender}")
             return
+        self._last_activity = time.monotonic()
         self._buffer.receive_data(data)
 
     def _on_message(self, msg: RoutableMessage) -> None:
@@ -477,6 +551,7 @@ async def _send(
             assert self.client is not None
             try:
                 await self.client.write_gatt_char(WRITE_UUID, payload, True)
+                self._last_activity = time.monotonic()
             # bleak-esphome converts an aioesphomeapi write timeout into a
             # builtin TimeoutError, not a BleakError, so catch both to keep the
             # GATT-write transport failure within TeslaFleetError.

tesla_fleet_api/tesla/vehicle/vehicles.py

@@ -5,7 +5,10 @@
 from cryptography.hazmat.primitives.asymmetric import ec
 
 from tesla_fleet_api.tesla.vehicle.signed import VehicleSigned
-from tesla_fleet_api.tesla.vehicle.bluetooth import VehicleBluetooth
+from tesla_fleet_api.tesla.vehicle.bluetooth import (
+    DEFAULT_KEEPALIVE_INTERVAL,
+    VehicleBluetooth,
+)
 from tesla_fleet_api.tesla.vehicle.fleet import VehicleFleet
 from tesla_fleet_api.tesla.vehicle.vehicle import Vehicle
 
@@ -41,14 +44,24 @@ def createSigned(self, vin: str) -> VehicleSigned[FleetParentT]:
         return vehicle
 
     def createBluetooth(
-        self, vin: str, verify_commands: bool = False
+        self,
+        vin: str,
+        verify_commands: bool = False,
+        keepalive_interval: float | None = DEFAULT_KEEPALIVE_INTERVAL,
     ) -> VehicleBluetooth[FleetParentT]:
         """Creates a bluetooth vehicle that uses command protocol.
 
         Set ``verify_commands`` to confirm supported mutating BLE command
         timeouts by reading the resulting state before surfacing the timeout.
+        ``keepalive_interval`` seconds of GATT idleness triggers a passive read
+        to hold the link open (``None``/``0`` disables).
         """
-        vehicle = self.Bluetooth(self._parent, vin, verify_commands=verify_commands)
+        vehicle = self.Bluetooth(
+            self._parent,
+            vin,
+            verify_commands=verify_commands,
+            keepalive_interval=keepalive_interval,
+        )
         self[vin] = vehicle
         return vehicle
 
@@ -76,28 +89,41 @@ def create(
         key: ec.EllipticCurvePrivateKey | None = None,
         device: BLEDevice | None = None,
         verify_commands: bool = False,
+        keepalive_interval: float | None = DEFAULT_KEEPALIVE_INTERVAL,
     ) -> VehicleBluetooth[BluetoothClientT]:
         """Creates a bluetooth vehicle that uses command protocol.
 
         Set ``verify_commands`` to confirm supported mutating BLE command
         timeouts by reading the resulting state before surfacing the timeout.
+        ``keepalive_interval`` seconds of GATT idleness triggers a passive read
+        to hold the link open (``None``/``0`` disables).
         """
-        return self.createBluetooth(vin, key, device, verify_commands)
+        return self.createBluetooth(
+            vin, key, device, verify_commands, keepalive_interval
+        )
 
     def createBluetooth(
         self,
         vin: str,
         key: ec.EllipticCurvePrivateKey | None = None,
         device: BLEDevice | None = None,
         verify_commands: bool = False,
+        keepalive_interval: float | None = DEFAULT_KEEPALIVE_INTERVAL,
     ) -> VehicleBluetooth[BluetoothClientT]:
         """Creates a bluetooth vehicle that uses command protocol.
 
         Set ``verify_commands`` to confirm supported mutating BLE command
         timeouts by reading the resulting state before surfacing the timeout.
+        ``keepalive_interval`` seconds of GATT idleness triggers a passive read
+        to hold the link open (``None``/``0`` disables).
         """
         vehicle = self.Bluetooth(
-            self._parent, vin, key, device, verify_commands=verify_commands
+            self._parent,
+            vin,
+            key,
+            device,
+            verify_commands=verify_commands,
+            keepalive_interval=keepalive_interval,
         )
         self[vin] = vehicle
         return vehicle

tesla_fleet_api/teslemetry/vehicle.py

@@ -303,6 +303,11 @@ def createSigned(self, vin: str) -> Any:
         """Creates a specific vehicle."""
         raise NotImplementedError("Teslemetry cannot use Fleet API directly")
 
-    def createBluetooth(self, vin: str, verify_commands: bool = False) -> Any:
-        """Creates a specific vehicle."""
+    def createBluetooth(
+        self,
+        vin: str,
+        verify_commands: bool = False,
+        keepalive_interval: float | None = None,
+    ) -> Any:
+        """Not supported; parameters match the Fleet API Bluetooth factory."""
         raise NotImplementedError("Teslemetry cannot use local Bluetooth")

tesla_fleet_api/tessie/vehicle.py

@@ -1199,6 +1199,11 @@ def createSigned(self, vin: str) -> Any:
         """Creates a specific vehicle."""
         raise NotImplementedError("Tessie cannot use Fleet API directly")
 
-    def createBluetooth(self, vin: str, verify_commands: bool = False) -> Any:
-        """Creates a specific vehicle."""
+    def createBluetooth(
+        self,
+        vin: str,
+        verify_commands: bool = False,
+        keepalive_interval: float | None = None,
+    ) -> Any:
+        """Not supported; parameters match the Fleet API Bluetooth factory."""
         raise NotImplementedError("Tessie cannot use local Bluetooth")