feat(vehicle): add BLE broadcast listeners

AGENTS.md

@@ -129,6 +129,7 @@ Source protobuf definitions live in `proto/`; generated Python files live in `te
 - **BLE mutating-command timeout is inconclusive - never assume "the write didn't land"**: this corrects an earlier version of this note, which observed `adjust_volume` write timeouts on one rig leaving the car unmutated and concluded a write timeout means the write never landed. Later live testing disproved that as a general rule: `door_unlock` and `door_lock` each raised a timeout yet both physically executed - a VCSEC state read after each confirmed the lock state had flipped. Reads (state readers, `wake_up()`'s effect) came back reliably all night; only mutating VCSEC/RKE actions showed this false-negative pattern, most likely because the vehicle doesn't reliably return an ack `_send()` observes within the timeout, not because the write failed. Treat `BluetoothUnconfirmedCommand` (a `BluetoothTimeout` subclass) from any mutating BLE command as **inconclusive, not failure**: snapshot state before acting, then verify the outcome with a follow-up state read whenever a mutation times out. Never blind-retry a non-idempotent command (toggles like `media_toggle_playback`, volume steps, schedule add/remove) on timeout alone - see the retry-double-execution entry below for why the library's own retry has the same exposure. The `adjust_volume`/`TimeoutAPIError` root cause from the original observation is still unexplained; it just isn't evidence that timed-out writes never land. VCSEC actuations now use the shorter `_actuation_timeout` when their terminal ack is lost.
 - **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`. Every `VehicleStatus` leaf field is a well-defined protobuf enum/int, so it 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`); anything not decoded into `VehicleStatus` (other VCSEC payloads like `CommandStatus`/whitelist/faults, and any future infotainment-domain broadcast - none observed today) has no typed listener and is 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`.
 - **`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

@@ -187,6 +187,11 @@ failures (including `bleak.exc.BleakError` and builtin `TimeoutError` from
 ESPHome proxies) and response-wait timeouts through the same library error
 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.
+
 ### Routing and Failover
 
 The `Router` class composes an ordered list of two-or-more backends that share a common method surface and dispatches each method call down the chain, automatically failing over on most errors. `VehicleRouter` and `EnergySiteRouter` are thin entity-specific subclasses. A common setup is a local `VehicleBluetooth` primary with a cloud fallback (e.g. a `TeslemetryVehicle`), so commands go over Bluetooth when the vehicle is reachable and route to the cloud otherwise:

docs/bluetooth_vehicles.md

@@ -472,6 +472,69 @@ endpoint, because requesting multiple endpoints together can exceed the
 vehicle's signed-command response-size cap and raise
 `TeslaFleetMessageFaultResponseSizeExceedsMTU`.
 
+## Broadcast Listeners
+
+The vehicle's VCSEC computer emits unsolicited `VehicleStatus` broadcasts on
+the same BLE notification subscription used for command replies, independent
+of any command you send. `VehicleBluetooth` fans these out to persistent
+per-field listeners, so you can receive vehicle-state changes passively
+instead of polling the state readers above.
+
+Every `VehicleStatus` leaf field is a well-defined protobuf enum or int, so
+each has its own typed listener method, similar in spirit to
+[python-teslemetry-stream](https://github.com/Teslemetry/python-teslemetry-stream)'s
+`listen_<Field>` surface:
+
+- `listen_vehicle_lock_state(callback)` - `VehicleLockState_E`
+- `listen_vehicle_sleep_status(callback)` - `VehicleSleepStatus_E`
+- `listen_user_presence(callback)` - `UserPresence_E`
+- `listen_front_driver_door(callback)` - `ClosureState_E`
+- `listen_front_passenger_door(callback)` - `ClosureState_E`
+- `listen_rear_driver_door(callback)` - `ClosureState_E`
+- `listen_rear_passenger_door(callback)` - `ClosureState_E`
+- `listen_front_trunk(callback)` - `ClosureState_E`
+- `listen_rear_trunk(callback)` - `ClosureState_E`
+- `listen_charge_port(callback)` - `ClosureState_E`
+- `listen_tonneau(callback)` - `ClosureState_E`
+- `listen_tonneau_percent_open(callback)` - `int`
+
+Each `listen_*` method takes a synchronous `callback(value)` and returns an
+`unsubscribe()` closure:
+
+```python
+def on_lock_state(state):
+    print(f"Lock state: {state}")
+
+unsubscribe = vehicle.listen_vehicle_lock_state(on_lock_state)
+...
+unsubscribe()
+```
+
+The door/trunk/charge-port/tonneau listeners and `listen_tonneau_percent_open`
+only fire on broadcasts that actually carry the corresponding submessage
+(`closureStatuses`/`detailedClosureStatus`) - not every status broadcast
+includes them. `listen_vehicle_lock_state`, `listen_vehicle_sleep_status`, and
+`listen_user_presence` fire on every `VehicleStatus` broadcast, since proto3
+gives no presence tracking for a scalar enum field.
+
+Anything not decoded into `VehicleStatus` - other VCSEC broadcast payloads
+(`CommandStatus`, whitelist events, faults) and any future
+infotainment-domain broadcast - has no typed listener surface. Use the untyped
+`listen_broadcast`, which delivers every raw unsolicited `RoutableMessage` for
+a given `Domain`, including decoded `VehicleStatus` broadcasts:
+
+```python
+from tesla_fleet_api.tesla.vehicle.proto.universal_message_pb2 import Domain
+
+unsubscribe = vehicle.listen_broadcast(Domain.DOMAIN_VEHICLE_SECURITY, print)
+```
+
+Listeners are plain Python callables registered on the `VehicleBluetooth`
+instance - they persist across reconnects and are torn down only by calling
+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.
+
 ## Media Commands
 
 `VehicleBluetooth` inherits the signed media commands from `Commands`, so media

tesla_fleet_api/tesla/vehicle/bluetooth.py

@@ -26,6 +26,7 @@
     TeslaFleetError,
     WhitelistOperationStatus,
 )
+from tesla_fleet_api.tesla.vehicle.broadcast import BroadcastListeners
 from tesla_fleet_api.tesla.vehicle.commands import (
     Commands,
     infotainment_command_name,
@@ -298,7 +299,9 @@ def _plan_auto_conditioning(action: VehicleAction) -> VerifyPlan | None:
 }
 
 
-class VehicleBluetooth(Commands[BluetoothParentT], Generic[BluetoothParentT]):
+class VehicleBluetooth(
+    BroadcastListeners, Commands[BluetoothParentT], Generic[BluetoothParentT]
+):
     """Class describing the Tesla Fleet API vehicle endpoints and commands for a specific vehicle with command signing.
 
     Callers can catch failures from this class with a single ``TeslaFleetError``,
@@ -382,6 +385,12 @@ class VehicleBluetooth(Commands[BluetoothParentT], Generic[BluetoothParentT]):
     per-command table (one source of truth), just applied to a broadcast frame
     or a follow-up read respectively.
 
+    The same unsolicited broadcast stream also feeds persistent listeners
+    inherited from ``BroadcastListeners``: typed ``listen_*`` methods for
+    decoded ``VehicleStatus`` fields and ``listen_broadcast`` for raw
+    per-domain broadcast messages. These listeners are local callables, persist
+    across reconnects, and are removed by the returned unsubscribe closure.
+
     ``raise_unconfirmed`` (default ``False``) is the orthogonal question of what
     to do once a ladder genuinely cannot resolve - the ack/broadcast wait
     (and, under ``"verify"``, the prover read) neither confirmed nor
@@ -489,6 +498,7 @@ def __init__(
             Domain.DOMAIN_INFOTAINMENT: asyncio.Queue(),
         }
         self._broadcast_watchers = {}
+        self._init_broadcast_listeners()
         self.device = device
         self._connect_lock = asyncio.Lock()
         self._buffer = ReassemblingBuffer(self._on_message)
@@ -655,9 +665,15 @@ def _on_message(self, msg: RoutableMessage) -> None:
 
         if msg.to_destination.routing_address != self._from_destination:
             LOGGER.debug("Ignoring broadcast message (not addressed to us)")
-            watcher = self._broadcast_watchers.get(msg.from_destination.domain)
+            domain = msg.from_destination.domain
+            watcher = self._broadcast_watchers.get(domain)
             if watcher is not None:
                 watcher(msg)
+            self._dispatch_domain_listeners(domain, msg)
+            if domain == Domain.DOMAIN_VEHICLE_SECURITY and self._status_listeners:
+                status = _decode_vcsec_status(msg)
+                if status is not None:
+                    self._dispatch_status_listeners(status)
             return
 
         queue = self._queues.get(msg.from_destination.domain)