feat(vehicle): verify BLE command timeouts by state

AGENTS.md

@@ -76,7 +76,7 @@ Commands (vehicle/commands.py) - protobuf-based signed command implementation (A
 `Vehicles` (vehicle/vehicles.py) is a `dict[str, Vehicle]` with factory methods:
 - `createFleet(vin)` → `VehicleFleet`
 - `createSigned(vin)` → `VehicleSigned`
-- `createBluetooth(vin)` → `VehicleBluetooth`
+- `createBluetooth(vin, verify_commands=False)` → `VehicleBluetooth`
 
 Teslemetry/Tessie override `Vehicles` with their own vehicle classes (`TeslemetryVehicle`, `TessieVehicle`) extending `VehicleFleet` with service-specific commands (e.g., `closure()`, `seat_heater()` for Teslemetry; `wake()`, `lock()` for Tessie).
 
@@ -127,6 +127,7 @@ Source protobuf definitions live in `proto/`; generated Python files live in `te
 - **`charge_standard()` rejects `already_standard`**: live-verified - calling it while `charge_state().charge_limit_soc` already equals `charge_limit_soc_std` gets `{"result": False, "reason": "already_standard"}` rather than a no-op success. Not a library bug; callers/tests exercising this command need the limit to actually differ from the std preset first (e.g. via `charge_max_range()` or `set_charge_limit()`).
 - **BLE media state-observability gotcha**: `MediaState.now_playing_artist/title` and all of `MediaDetailState` (`now_playing_album/station/source_string/elapsed/duration`) were observed empty/zero on the test car while Spotify was actively `Playing` at nonzero volume - these legacy fields are apparently only populated for certain sources (e.g. USB/Bluetooth), not Spotify. Don't assume `media_next_track`/`media_prev_track`/`media_next_fav`/`media_prev_fav` are state-observable via these readers; verify by ACK (`{"result": True}`) and pair with the inverse command when the fingerprint doesn't change. `audio_volume`/`media_playback_status` (for `adjust_volume`/`media_volume_up`/`media_volume_down`/`media_toggle_playback`) were populated correctly and are reliable provers.
 - **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 `BluetoothTimeout` 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 a `BluetoothTimeout` 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.
+- **Opt-in `verify_commands` resolves the inconclusive mutating-timeout inside `VehicleBluetooth`**: constructor knob `verify_commands` (default off, `bluetooth.py`; threaded through `Vehicles`/`VehiclesBluetooth.createBluetooth`) that, on a `BluetoothTimeout` from a mutating command whose expected post-state is derivable from its args, reads the mapped prover state and returns a normal success dict if it matches or re-raises the timeout if not. Verify-on-timeout only (not verify-always): the post-PR-59 happy path returns on the terminal ack, so the prover read is paid only in the ambiguous case - "faster and still reliable". Intercepts at the `_sendVehicleSecurity`/`_sendInfotainment` seam (each called exactly once per public mutation; reads go through `_get*`, so a verification read never recurses into verification). Lives here rather than upstream because BLE is the only transport with the false-negative-ack problem; with it enabled, `VehicleRouter`'s blind BLE->cloud failover no longer double-executes non-idempotent commands (a verified-executed command returns success so no failover fires; a verified-not-executed one raises so failover is genuinely safe - Router needs no changes). Mapping tables and predicates are in `bluetooth.py` (`_vcsec_verify_plan`/`_INFOTAINMENT_VERIFY_PLANS`); only clearly-derivable, absolute commands are covered (lock/unlock, `set_charge_limit`, `set_charging_amps`, `adjust_volume` absolute, `set_temps`, `auto_conditioning_start/stop`). True toggles, relative volume steps, and ack-only actions have no plan and re-raise unchanged. The VCSEC lock prover reads while asleep; INFO provers need the car awake and never wake it - an asleep-car read failure re-raises the original timeout. See `docs/bluetooth_vehicles.md` for the user-facing table.
 - **`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 a `BluetoothTimeout` is still only an inconclusive wake signal, not command failure. Call it best-effort (catch and ignore timeout) and 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.
 - **The signed-command retry in `Commands._command` can double-execute a mutating command**: on an `OPERATIONSTATUS_WAIT` reply or an `INCORRECT_EPOCH`/`INVALID_TOKEN` fault, `_command` (`commands.py`) re-signs and re-sends the identical command, bounded at 3 attempts then a clean `{"result": False, "reason": "Too many retries"}` - the cap itself is safe and doesn't loop. Live-verified deterministically: a WAIT-then-OK sequence produces 2 physical sends of the same command on the wire. Combined with the mutating-timeout-is-inconclusive gotcha above (a command can execute despite a WAIT/fault reply), this retry is a latent double-apply window. Harmless for a naturally idempotent command (lock/unlock), a real correctness risk for toggles and step commands (`media_toggle_playback`, `media_volume_up`/`down`, schedule add/remove) - verify those by absolute state after the call, never by counting invocations or trusting the retry to be safe.
 - **`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.

README.md

@@ -189,7 +189,7 @@ async def main():
         # Primary: local Bluetooth
         tesla_bluetooth = TeslaBluetooth()
         await tesla_bluetooth.get_private_key("path/to/private_key.pem")
-        primary = tesla_bluetooth.vehicles.create("<vin>")
+        primary = tesla_bluetooth.vehicles.create("<vin>", verify_commands=True)
 
         # Secondary (fallback): Teslemetry cloud
         teslemetry = Teslemetry(access_token="<access_token>", session=session)
@@ -220,7 +220,7 @@ await router.set_operation(...)  # local first, cloud on failure
 
 `Router`, `VehicleRouter`, and `EnergySiteRouter` are all importable from `tesla_fleet_api.router` (and, for backward compatibility, from `tesla_fleet_api.tesla`).
 
-> **Warning:** Because a failed call is replayed on the next backend, a non-idempotent command (e.g. `honk_horn`, `actuate_trunk`, `door_unlock`, `charge_start`) that fails _mid-flight_ — after a backend may have already partially applied it — can be **double-executed** (or executed more than once across a longer chain) when it is retried on the next backend. This is a deliberate tradeoff of per-command failover. Callers needing exactly-once semantics for such commands should gate dispatch with an explicit `health` check or call the underlying backends directly.
+> **Warning:** Because a failed call is replayed on the next backend, a non-idempotent command (e.g. `honk_horn`, `actuate_trunk`, `door_unlock`, `charge_start`) that fails _mid-flight_ — after a backend may have already partially applied it — can be **double-executed** (or executed more than once across a longer chain) when it is retried on the next backend. This is a deliberate tradeoff of per-command failover. When the primary is `VehicleBluetooth`, pass `verify_commands=True` to resolve supported mutating command timeouts by state before failover; callers needing exactly-once semantics for other commands should gate dispatch with an explicit `health` check or call the underlying backends directly.
 >
 > Dispatch is implemented via `__getattr__`, which does not proxy dunder methods, so `async with Router(...)` does **not** manage a backend's BLE connection lifecycle (`__aenter__`/`__aexit__`). Commands still auto-connect on send; for explicit connect/disconnect reach through `router.primary` (or `router.backends`).
 

docs/bluetooth_vehicles.md

@@ -127,6 +127,41 @@ inside the library can also re-send an identical command after a WAIT status or
 epoch/token fault, so verify by absolute state rather than by counting command
 attempts.
 
+### Opt-in state verification (`verify_commands`)
+
+Construct a `VehicleBluetooth` with `verify_commands=True` (also accepted by
+`vehicles.create(...)` / `vehicles.createBluetooth(...)`, default off) to have
+the class resolve that ambiguity for you. Verification runs only on a timeout,
+so the normal path - which returns as soon as the terminal ack arrives - keeps
+its speed and pays no extra read; the prover read happens only in the ambiguous
+case.
+
+When a mutating command times out and its expected post-state can be derived
+from its own arguments, the same held connection reads the mapped prover state
+and either returns a normal success result (`{"response": {"result": True,
+"reason": ""}}`) when the state matches or re-raises the `BluetoothTimeout` when
+it does not. The read rides the existing connection and never wakes the vehicle;
+if an infotainment prover cannot be read because the car is asleep, the original
+timeout is re-raised.
+
+Commands whose outcome cannot be derived or read - true toggles, relative volume
+steps, and ack-only actions such as `flash_lights()` or `trigger_homelink()` -
+re-raise the timeout unchanged, exactly as with verification off. Currently
+verified commands and their provers:
+
+| Command | Prover | Confirmed when |
+| --- | --- | --- |
+| `door_lock()` / `door_unlock()` | `vehicle_state()` | lock state matches |
+| `set_charge_limit(percent)` | `charge_state()` | `charge_limit_soc` matches |
+| `set_charging_amps(amps)` | `charge_state()` | `charging_amps` matches |
+| `adjust_volume(volume)` | `media_state()` | `audio_volume` matches |
+| `set_temps(driver, passenger)` | `climate_state()` | both temp settings match |
+| `auto_conditioning_start()` / `auto_conditioning_stop()` | `climate_state()` | `is_climate_on` matches |
+
+The VCSEC lock prover is readable while the vehicle is asleep; the infotainment
+provers require the vehicle awake. This feature does not change behavior for any
+command not in the table, and it is inert unless `verify_commands=True`.
+
 ## Climate Commands
 
 Bluetooth vehicles support the same signed climate command methods as

tesla_fleet_api/tesla/vehicle/bluetooth.py

@@ -20,6 +20,7 @@
     WHITELIST_OPERATION_STATUS,
     BluetoothTimeout,
     BluetoothTransportError,
+    TeslaFleetError,
     WhitelistOperationStatus,
 )
 from tesla_fleet_api.tesla.vehicle.commands import Commands
@@ -60,6 +61,7 @@
     PublicKey,
     RKEAction_E,
     UnsignedMessage,
+    VehicleLockState_E,
     VehicleStatus,
     WhitelistOperation,
 )
@@ -194,6 +196,78 @@ def discard_packet(self):
         self.expected_length = None
 
 
+# Optional post-timeout command verification.
+#
+# A lost ack from a mutating BLE command is inconclusive: the vehicle may have
+# executed it anyway. When ``verify_commands`` is on, a timed-out mutation whose
+# outcome can be derived from its own arguments is confirmed by reading the
+# mapped state instead of surfacing the ambiguous timeout. A verify "plan" pairs
+# the state reader to call with a predicate that checks the observed state
+# against the requested value. Commands absent from these tables - true toggles,
+# relative steps, and ack-only actions whose outcome cannot be derived from the
+# request - have no plan and re-raise the timeout unchanged.
+VerifyPlan = tuple[str, Callable[[Any], bool]]
+
+
+def _vcsec_verify_plan(command: UnsignedMessage) -> VerifyPlan | None:
+    """Derive a VCSEC verify plan (read ``vehicle_state``) from a lost actuation."""
+    if command.WhichOneof("sub_message") != "RKEAction":
+        return None
+    if command.RKEAction == RKEAction_E.RKE_ACTION_LOCK:
+        return "vehicle_state", (
+            lambda s: s.vehicleLockState == VehicleLockState_E.VEHICLELOCKSTATE_LOCKED
+        )
+    if command.RKEAction == RKEAction_E.RKE_ACTION_UNLOCK:
+        return "vehicle_state", (
+            lambda s: s.vehicleLockState == VehicleLockState_E.VEHICLELOCKSTATE_UNLOCKED
+        )
+    return None
+
+
+def _plan_charge_limit(action: VehicleAction) -> VerifyPlan | None:
+    percent = action.chargingSetLimitAction.percent
+    return "charge_state", lambda s: s.charge_limit_soc == percent
+
+
+def _plan_charging_amps(action: VehicleAction) -> VerifyPlan | None:
+    amps = action.setChargingAmpsAction.charging_amps
+    return "charge_state", lambda s: s.charging_amps == amps
+
+
+def _plan_volume(action: VehicleAction) -> VerifyPlan | None:
+    # A relative volume step is not derivable without a pre-read, so only the
+    # absolute set is verifiable.
+    if action.mediaUpdateVolume.WhichOneof("media_volume") != "volume_absolute_float":
+        return None
+    target = action.mediaUpdateVolume.volume_absolute_float
+    return "media_state", lambda s: s.audio_volume == target
+
+
+def _plan_temps(action: VehicleAction) -> VerifyPlan | None:
+    driver = action.hvacTemperatureAdjustmentAction.driver_temp_celsius
+    passenger = action.hvacTemperatureAdjustmentAction.passenger_temp_celsius
+    return "climate_state", (
+        lambda s: (
+            s.driver_temp_setting == driver and s.passenger_temp_setting == passenger
+        )
+    )
+
+
+def _plan_auto_conditioning(action: VehicleAction) -> VerifyPlan | None:
+    power_on = action.hvacAutoAction.power_on
+    return "climate_state", lambda s: s.is_climate_on == power_on
+
+
+# Keyed by the ``VehicleAction`` oneof field an infotainment mutation sets.
+_INFOTAINMENT_VERIFY_PLANS: dict[str, Callable[[VehicleAction], VerifyPlan | None]] = {
+    "chargingSetLimitAction": _plan_charge_limit,
+    "setChargingAmpsAction": _plan_charging_amps,
+    "mediaUpdateVolume": _plan_volume,
+    "hvacTemperatureAdjustmentAction": _plan_temps,
+    "hvacAutoAction": _plan_auto_conditioning,
+}
+
+
 class VehicleBluetooth(Commands[BluetoothParentT], Generic[BluetoothParentT]):
     """Class describing the Tesla Fleet API vehicle endpoints and commands for a specific vehicle with command signing.
 
@@ -211,9 +285,19 @@ class VehicleBluetooth(Commands[BluetoothParentT], Generic[BluetoothParentT]):
     blind-retry a non-idempotent command (toggles, volume steps, schedule
     add/remove) on a timeout alone. The inherited WAIT/fault retry
     (``Commands._command``) can also re-send an already-executed command.
+
+    Pass ``verify_commands=True`` to resolve that ambiguity inside this class:
+    on a timeout from a mutating command whose expected post-state is derivable
+    from its arguments, the same held connection reads the mapped prover state
+    and either returns a normal success result (the command executed) or
+    re-raises the ``BluetoothTimeout`` (it did not). Commands whose outcome
+    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).
     """
 
     ble_name: str
+    verify_commands: bool
     device: BLEDevice | None = None
     client: BleakClient | None = None
     _queues: dict[Domain, asyncio.Queue[RoutableMessage]]
@@ -232,8 +316,10 @@ def __init__(
         vin: str,
         key: ec.EllipticCurvePrivateKey | None = None,
         device: BLEDevice | None = None,
+        verify_commands: bool = False,
     ):
         super().__init__(parent, vin, key)
+        self.verify_commands = verify_commands
         self.ble_name = "S" + hashlib.sha1(vin.encode("utf-8")).hexdigest()[:16] + "C"
         self._queues = {
             Domain.DOMAIN_VEHICLE_SECURITY: asyncio.Queue(),
@@ -582,6 +668,49 @@ async def _command(
             domain, command, attempt, expects_data=expects_data
         )
 
+    async def _sendVehicleSecurity(self, command: UnsignedMessage) -> dict[str, Any]:
+        """Send a VCSEC actuation, optionally resolving a lost-ack timeout by state."""
+        if not self.verify_commands:
+            return await super()._sendVehicleSecurity(command)
+        try:
+            return await super()._sendVehicleSecurity(command)
+        except BluetoothTimeout as timeout:
+            return await self._resolve_timeout(_vcsec_verify_plan(command), timeout)
+
+    async def _sendInfotainment(self, command: Action) -> dict[str, Any]:
+        """Send an infotainment command, optionally resolving a lost-ack timeout by state."""
+        if not self.verify_commands:
+            return await super()._sendInfotainment(command)
+        try:
+            return await super()._sendInfotainment(command)
+        except BluetoothTimeout as timeout:
+            resolver = _INFOTAINMENT_VERIFY_PLANS.get(
+                command.vehicleAction.WhichOneof("vehicle_action_msg")
+            )
+            plan = resolver(command.vehicleAction) if resolver else None
+            return await self._resolve_timeout(plan, timeout)
+
+    async def _resolve_timeout(
+        self, plan: VerifyPlan | None, timeout: BluetoothTimeout
+    ) -> dict[str, Any]:
+        """Confirm a timed-out mutation by state, or re-raise if unverifiable.
+
+        The prover read rides the same held connection. An INFO-domain prover
+        needs the vehicle awake; if the read cannot complete (e.g. the car is
+        asleep) it surfaces a ``TeslaFleetError`` and the original timeout is
+        re-raised rather than waking the car just to verify.
+        """
+        if plan is None:
+            raise timeout
+        reader_name, executed = plan
+        try:
+            state = await getattr(self, reader_name)()
+        except TeslaFleetError:
+            raise timeout
+        if executed(state):
+            return {"response": {"result": True, "reason": ""}}
+        raise timeout
+
     async def pair(
         self,
         role: Role = Role.ROLE_OWNER,