feat: add BLE connection status events

AGENTS.md

@@ -132,6 +132,7 @@ Keep the `tesla-protocol` floor at `>=0.5.0`; earlier releases have generated `.
 - **The BLE mutating-command confirmation ladder is one `confirmation` enum + one `raise_unconfirmed` bool**: `VehicleBluetooth.confirmation` (`"optimistic" | "ack" | "verify"`, default `"ack"`; threaded through `Vehicles`/`VehiclesBluetooth.create*`) picks how many of write → ack-or-broadcast wait → state-read confirmation run; `raise_unconfirmed` (default `False`) picks what happens when the ladder still can't tell. `"optimistic"` short-circuits `_sendVehicleSecurity`/`_sendInfotainment` (`bluetooth.py`) to `_send_optimistic()`, which signs and writes but never waits for any reply - a provably pre-submission write failure still raises `BluetoothTransportError` unconditionally, but a submitted-then-ambiguous write (see the write-delivery-certainty entry below) follows `raise_unconfirmed` like every other rung instead of being consulted as "nothing else." `"verify"` adds a post-timeout state-read rung: on an unresolved ack/broadcast wait, `_resolve_timeout()` reads the mapped prover state (`_vcsec_verify_plan`/`_INFOTAINMENT_VERIFY_PLANS` in `bluetooth.py`; only clearly-derivable absolute commands are covered - lock/unlock, `set_charge_limit`, `set_charging_amps`, `adjust_volume` absolute, `set_temps`, `auto_conditioning_start/stop`) and returns success on a match, raises `BluetoothCommandFailed` on a proven mismatch, or returns `None` (still unresolved) if the read itself couldn't complete - `None` falls through to `raise_unconfirmed`. Commands with no plan (true toggles, relative steps, ack-only actions) always fall through regardless of `confirmation`. The legacy `optimistic`/`verify_commands` boolean surface is deprecated: both warn (`DeprecationWarning`) and map onto `confirmation` (a positional bool in the `confirmation` slot is treated as old `verify_commands`; dominance order preserved - `optimistic=True` wins if both are set), and remain as read-only properties (`confirmation == "optimistic"`/`"verify"`) for existing readers. See `docs/bluetooth_vehicles.md` for the user-facing table and defaults.
 - **Broadcast-as-confirmation races the ack wait for lock/unlock**: the vehicle keeps emitting unsolicited VCSEC status broadcasts on the same notification subscription even when it emits no addressed ack for a lock/unlock actuation - live-verified at scale (see the timeout-rate study referenced from `docs/bluetooth_vehicles.md`). `_send`'s `confirm_broadcast` param (threaded through `Commands._command`/`_sendVehicleSecurity`, ignored by the Fleet-signed transport) arms a per-domain watcher in `_on_message` (`_broadcast_watchers`, `bluetooth.py`) that decodes broadcast frames via `_decode_vcsec_status` and races them against the addressed-reply wait in `_await_response_or_broadcast`; first to satisfy the plan's predicate wins, and only the addressed-reply path can raise a car-side rejection. A mismatching broadcast doesn't fail fast (it's appended to `mismatches`, not resolved) since a later broadcast in the same window could still confirm success - but if the whole window elapses with a mismatch as the last word and nothing else confirming, `_await_response_or_broadcast` raises `BluetoothCommandFailed` instead of the ambiguous timeout. This reuses the exact same `_vcsec_verify_plan` predicate as the `"verify"` rung above (one source of truth), applied to a broadcast's decoded `VehicleStatus` instead of a follow-up read; it currently covers only lock/unlock, the one VCSEC actuation with an observed status broadcast. Low-level race tests drive the real (unmocked) `_send` state machine directly - see `tests/test_ble_broadcast_confirmation.py`.
 - **Persistent broadcast listeners (`tesla_fleet_api/tesla/vehicle/broadcast.py`)**: `VehicleBluetooth` fans the same VCSEC status broadcasts out to long-lived per-field listeners, not just the one-shot confirmation ladder above - one source of broadcast truth, dispatched from the same `_on_message`. Each modeled `VehicleStatus` leaf field gets a typed `listen_<field>` method (`listen_vehicle_lock_state`, `listen_vehicle_sleep_status`, `listen_user_presence`, the 8 door/trunk/charge-port/tonneau closure listeners, `listen_tonneau_percent_open`); `uiDesire`/`gear` (added to `VehicleStatus` by the app-recovered proto sync) have no typed listener yet, and anything not decoded into `VehicleStatus` (other VCSEC payloads like `CommandStatus`/whitelist/faults, and any future infotainment-domain broadcast - none observed today) also has none - all of these are covered by the untyped `listen_broadcast(domain, callback)`, which receives every raw unsolicited broadcast for that domain including `VehicleStatus`. Closure/tonneau-percent listeners gate on `HasField` since those are submessages with real proto3 presence tracking; the three scalar enum fields (`vehicleLockState`/`vehicleSleepStatus`/`userPresence`) have none, so they fire on every status broadcast rather than only on change. Each `listen_*` returns an `unsubscribe()` closure; registries live for the `VehicleBluetooth` instance's lifetime and are unaffected by reconnects, matching `_queues`. Listener callback exceptions are logged and isolated from later listeners/message routing, except `KeyboardInterrupt`/`SystemExit`. Not consumed by Home Assistant yet - HA gets state via Teslemetry streaming already. See `docs/bluetooth_vehicles.md` and `tests/test_ble_broadcast_listeners.py`.
+- **Connection-status listener**: `VehicleBluetooth.listen_connection_status()` reports BLE session transitions, including unexpected transport loss; the authoritative contract is in `docs/bluetooth_vehicles.md#connection-status-events` and its regression coverage is in `tests/test_ble_connection_status.py`.
 - **`BluetoothUnconfirmedCommand` vs `BluetoothCommandFailed` (`exceptions.py`), and how `Router` treats each**: `_sendVehicleSecurity`/`_sendInfotainment` (`bluetooth.py`, the mutating-command seam, unconditionally) wrap a caught `BluetoothTimeout` into `BluetoothUnconfirmedCommand` when the ladder is genuinely unresolved - either the write succeeded but the ack/broadcast was lost, or the write entered backend I/O and failed with delivery unprovable, so the vehicle may have executed the command. With default `raise_unconfirmed=False` that unresolved outcome returns best-effort success; with `raise_unconfirmed=True` the `BluetoothUnconfirmedCommand` reaches the caller. `BluetoothCommandFailed` is the other, distinct outcome: a state check (the `"verify"` rung's read, or a mismatching broadcast still standing at window-end) actively *proved* the command did not apply - it deliberately does **not** subclass `BluetoothTimeout`/`BluetoothUnconfirmedCommand`. `Router._dispatch` (`router/base.py`) special-cases only `BluetoothUnconfirmedCommand` to skip its normal per-command failover and re-raise immediately - replaying an already-possibly-executed command risks double-execution. `BluetoothCommandFailed` carries no such risk (the command is proven not to have applied), so it falls through `Router`'s ordinary `except (Exception, TeslaFleetError)` clause and fails over like any other error - no `Router` code change was needed for this, only the exception type's placement outside the `BluetoothTimeout` hierarchy. A plain read (`_getVehicleSecurity`/`_getInfotainment`) still raises unadorned `BluetoothTimeout` on the same kind of wait timeout, since a read has no side effect to be unconfirmed about, and a *provably pre-submission* write failure still raises the unrelated `BluetoothTransportError` (see the write-delivery-certainty entry below - a submitted-then-ambiguous write is not this case).
 - **Write-delivery certainty splits `BluetoothTransportError` from `BluetoothTimeout` at the GATT write in `_send`**: `write_gatt_char` failures are not uniformly `BluetoothTransportError`. `BleakCharacteristicNotFoundError` (bleak resolves `WRITE_UUID` synchronously, before any backend I/O) is the only case provably pre-submission, so it alone stays `BluetoothTransportError` and is safe for `Router` to retry. Every other `BleakError`/`TimeoutError` from that call happens inside backend I/O (D-Bus/CoreBluetooth/an ESPHome proxy) where delivery can't be proven either way - field data measured 2 of 3 such write failures had already executed on the vehicle - so `_send` instead races any already-armed broadcast watcher for the rest of the window (a matching broadcast can still confirm success despite the failed write) and, failing that, raises plain `BluetoothTimeout`. Because `BluetoothUnconfirmedCommand` subclasses `BluetoothTimeout`, this lands in the exact same `except BluetoothTimeout` ladder in `_sendVehicleSecurity`/`_sendInfotainment` as a lost post-write ack, with no separate exception type needed; `_send_optimistic` gets the equivalent treatment explicitly since it bypasses that ladder (see above). A read is unaffected - `_getVehicleSecurity`/`_getInfotainment` don't override the ladder, so a write-ambiguous read propagates plain `BluetoothTimeout` and `Router` fails over on it normally, which is safe since a read has no double-execution risk. Tests: `tests/test_ble_send_transport.py` (`SendTransportErrorTests`), `tests/test_ble_broadcast_confirmation.py` (`WriteFailureBroadcastRaceTests`), `tests/test_ble_write_timeout_router.py`.
 - **`wake_up()` is best-effort; confirm readiness with an INFO read**: `wake_up()` is a VCSEC actuation, so a terminal ack now returns promptly when observed, but an unresolved wake remains only an inconclusive wake signal, not command failure (`BluetoothUnconfirmedCommand` when `raise_unconfirmed=True`, best-effort success by default). Confirm readiness by retrying a cheap INFO read instead (see the boot-delay gotcha above for why the first INFO read still needs its own retry/backoff). Hold one connection across a whole batch of related commands rather than reconnecting between each - reconnecting costs ~123% more per operation with no demonstrated wake-preservation benefit from the connection alone.

README.md

@@ -201,7 +201,10 @@ hierarchy.
 `VehicleBluetooth` can also register persistent BLE broadcast listeners for
 unsolicited VCSEC `VehicleStatus` updates. Use typed `listen_*` helpers for the
 decoded vehicle-status fields, or `listen_broadcast(domain, callback)` for raw
-per-domain broadcast messages.
+per-domain broadcast messages. Use `listen_connection_status(callback)` for
+`True`/`False` BLE session transition notifications. See
+[Bluetooth for Vehicles](docs/bluetooth_vehicles.md#connection-status-events)
+for the connection-event contract.
 
 ### Routing and Failover
 

docs/bluetooth_vehicles.md

@@ -556,6 +556,30 @@ the `unsubscribe()` closure returned at registration.
 Callback exceptions are logged and do not stop later listeners or normal
 message routing; `KeyboardInterrupt` and `SystemExit` still propagate.
 
+### Connection-status events
+
+Use `listen_connection_status(callback)` to receive BLE session transitions
+without polling `is_connected`. The synchronous callback receives `True` after
+`connect()` has successfully subscribed to GATT notifications and `False` when
+that session is lost, whether through `disconnect()` or an unexpected transport
+drop detected by bleak during an operation:
+
+```python
+def on_connection_status(connected: bool):
+    print(f"BLE connected: {connected}")
+
+unsubscribe = vehicle.listen_connection_status(on_connection_status)
+...
+unsubscribe()
+```
+
+Only actual transitions are emitted, so redundant `connect()`/`disconnect()`
+calls and reconnect loops do not duplicate an unchanged state. Registration
+persists across reconnects until the returned, idempotent `unsubscribe()`
+closure is called. As with broadcast listeners, one callback raising an
+exception is logged and does not prevent later callbacks from running;
+`KeyboardInterrupt` and `SystemExit` still propagate.
+
 ## Media Commands
 
 `VehicleBluetooth` inherits the signed media commands from `Commands`, so media

tesla_fleet_api.egg-info/SOURCES.txt

@@ -59,6 +59,7 @@ tests/test_ble_charging_utility_commands.py
 tests/test_ble_climate_commands.py
 tests/test_ble_command_verification.py
 tests/test_ble_confirmation_mode.py
+tests/test_ble_connection_status.py
 tests/test_ble_connectivity_diagnostics_commands.py
 tests/test_ble_destructive_commands.py
 tests/test_ble_expects_data.py

tesla_fleet_api/tesla/vehicle/bluetooth.py

@@ -27,7 +27,7 @@
     TeslaFleetError,
     WhitelistOperationStatus,
 )
-from tesla_fleet_api.tesla.vehicle.broadcast import BroadcastListeners
+from tesla_fleet_api.tesla.vehicle.broadcast import BroadcastListeners, Unsubscribe
 from tesla_fleet_api.tesla.vehicle.commands import (
     Commands,
     infotainment_command_name,
@@ -496,6 +496,8 @@ class VehicleBluetooth(
     _keepalive_timeout: float = 2
     _keepalive_task: asyncio.Task[None] | None = None
     _last_activity: float = 0.0
+    _connected: bool = False
+    _connection_listeners: list[Callable[[bool], None]]
 
     def __init__(
         self,
@@ -559,6 +561,7 @@ def __init__(
         self._stream_sinks = {}
         self._retired_streams = deque(maxlen=_STREAM_TOMBSTONE_MAXSIZE)
         self._init_broadcast_listeners()
+        self._connection_listeners = []
         self.device = device
         self._connect_lock = asyncio.Lock()
         self._buffer = ReassemblingBuffer(self._on_message)
@@ -615,12 +618,14 @@ async def connect(self, max_attempts: int = MAX_CONNECT_ATTEMPTS) -> None:
                 BleakClient,
                 self.device,
                 self.vin,
+                disconnected_callback=self._on_ble_disconnected,
                 max_attempts=max_attempts,
                 # ble_device_callback=self.get_device,
                 services=[SERVICE_UUID],
             )
             await self.client.start_notify(READ_UUID, self._on_notify)
             await self._start_keepalive()
+            self._set_connected(True)
         # 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.
@@ -640,8 +645,40 @@ async def disconnect(self) -> bool:
         if not self.client:
             return False
         await self.client.disconnect()
+        self._set_connected(False)
         return True
 
+    def listen_connection_status(self, callback: Callable[[bool], None]) -> Unsubscribe:
+        """Listen for BLE session connect/disconnect events.
+
+        Fires ``True`` once a connection is fully established (GATT
+        notifications subscribed) and ``False`` once that session is lost -
+        cleanly via ``disconnect()`` or unexpectedly via bleak's own
+        ``disconnected_callback``, including a loss detected mid-operation.
+        Only genuine state transitions fire; a redundant ``connect()``/
+        ``disconnect()`` call, or a reconnect loop, does not re-fire while the
+        state hasn't actually changed.
+        """
+        return self._register(self._connection_listeners, callback)
+
+    def _set_connected(self, connected: bool) -> None:
+        if self._connected == connected:
+            return
+        self._connected = connected
+        for callback in list(self._connection_listeners):
+            self._dispatch_callback(callback, connected)
+
+    def _on_ble_disconnected(self, client: BleakClient) -> None:
+        """bleak's own disconnect callback - fires on any real session loss.
+
+        Covers both a clean ``disconnect()`` and an unexpected drop detected
+        mid-operation; ``_set_connected`` collapses either into at most one
+        ``False`` dispatch.
+        """
+        if client is not self.client:
+            return
+        self._set_connected(False)
+
     async def connect_if_needed(self, max_attempts: int = MAX_CONNECT_ATTEMPTS) -> None:
         """Connect to the Tesla BLE device if not already connected."""
         async with self._connect_lock: