fix(ble): harden BLE transport routing and reassembly

AGENTS.md

@@ -116,6 +116,7 @@ Source protobuf definitions live in `proto/`; generated Python files live in `te
 - **Seat indexing gotcha**: two distinct seat enums with different conventions. `Seat` is **0-indexed** (`FRONT_LEFT=0`) and is for the manual seat heater/cooler paths (`remote_seat_heater_request`, `remote_seat_cooler_request`). `AutoSeat` is **1-indexed** (`FRONT_LEFT=1`, `FRONT_RIGHT=2`) and is the correct type for `remote_auto_seat_climate_request` on **both** backends — its values equal Tesla's REST wire values and the proto `AutoSeatPosition_*` enum. Don't mix them; passing a `Seat` to the auto-climate command is off-by-one (issue #11).
 - **Naming**: camelCase for class instance attributes that mirror API structure (`energySites`, `createFleet`). Snake_case for method names that are API endpoints.
 - **BLE discovery gotcha**: a Tesla vehicle advertises no 128-bit service UUID pre-connect — only its VIN-derived local name (`^S[a-f0-9]{16}[CDRP]$`), and only in the scan response, not the `ADV_IND`. `SERVICE_UUID` (`tesla_fleet_api/tesla/vehicle/bluetooth.py`) exists only as a GATT service after connecting. Never pass `service_uuids=[SERVICE_UUID]` as a `BleakScanner` discovery-time filter — it hides the vehicle on a direct BlueZ adapter (an ESPHome proxy doesn't enforce that filter the same way, which can mask the bug in testing). Scan unfiltered with active scanning and match by name; keep `SERVICE_UUID` for post-connect GATT use only.
+- **BLE domain-routing gotcha**: `Domain` (`proto/universal_message.proto`) has more values (`DOMAIN_BROADCAST`, `DOMAIN_AUTHD`, ...) than `VehicleBluetooth._queues` has keys (only `DOMAIN_VEHICLE_SECURITY`/`DOMAIN_INFOTAINMENT`). `_on_message` (`tesla_fleet_api/tesla/vehicle/bluetooth.py`) must look up `_queues` with `.get()` and drop unrecognized domains rather than indexing directly — indexing raises `KeyError` inside the `ReassemblingBuffer` callback, aborting reassembly of any further already-buffered messages in that notification.
 
 ## Maintaining this file
 

tesla_fleet_api/tesla/vehicle/bluetooth.py

@@ -96,17 +96,19 @@ def prependLength(message: bytes) -> bytearray:
 
 class ReassemblingBuffer:
     """
-    Reassembles bytearray streams where the first two bytes indicate the length of the message.
-    Handles potential packet corruption by discarding *entire* packets and retrying.
-    Uses a callback to process parsed messages.
+    Reassembles BLE notification chunks into length-prefixed RoutableMessages.
+
+    Each message starts with a 2-byte length. One notification can contain part
+    of a message, exactly one message, or multiple messages. If a message cannot
+    be decoded, the buffer drops the current physical packet and resynchronizes
+    at the next recorded packet boundary.
     """
 
     def __init__(self, callback: Callable[[RoutableMessage], None]):
         """
         Initializes the buffer.
 
         Args:
-            message_type: The protobuf message type (e.g., RoutableMessage) to parse the assembled data.
             callback: A function that will be called with each parsed message.
         """
         self.buffer: bytearray = bytearray()
@@ -116,7 +118,7 @@ def __init__(self, callback: Callable[[RoutableMessage], None]):
 
     def receive_data(self, data: bytearray):
         """
-        Receives a chunk of bytearray data and attempts to assemble a complete message.
+        Receive one BLE notification chunk and emit any complete messages.
 
         Args:
             data: The received bytearray data.
@@ -181,6 +183,7 @@ class VehicleBluetooth(Commands[BluetoothParentT], Generic[BluetoothParentT]):
     _ekey: ec.EllipticCurvePublicKey
     _buffer: ReassemblingBuffer
     _auth_method = "aes"
+    _ack_followup_timeout: float = 2
 
     def __init__(
         self,
@@ -282,14 +285,24 @@ def _on_notify(self, sender: BleakGATTCharacteristic, data: bytearray) -> None:
         self._buffer.receive_data(data)
 
     def _on_message(self, msg: RoutableMessage) -> None:
-        """Receive messages from the Tesla BLE data."""
+        """Route addressed BLE replies into the per-domain response queue."""
 
         if msg.to_destination.routing_address != self._from_destination:
             LOGGER.debug("Ignoring broadcast message (not addressed to us)")
             return
 
+        queue = self._queues.get(msg.from_destination.domain)
+        if queue is None:
+            # Domain enum has values (e.g. DOMAIN_BROADCAST, DOMAIN_AUTHD) with
+            # no session/queue of our own; indexing _queues directly would
+            # raise KeyError and abort the reassembly loop mid-buffer.
+            LOGGER.debug(
+                f"Ignoring message from unhandled domain {msg.from_destination.domain}"
+            )
+            return
+
         LOGGER.debug(f"Received response: {msg}")
-        self._queues[msg.from_destination.domain].put_nowait(msg)
+        queue.put_nowait(msg)
 
     async def _send(
         self, msg: RoutableMessage, requires: str, timeout: int = 5
@@ -332,7 +345,7 @@ async def _send(
                                 "Received ACK for our request, waiting briefly for data follow-up"
                             )
                             try:
-                                async with asyncio.timeout(2):
+                                async with asyncio.timeout(self._ack_followup_timeout):
                                     while True:
                                         resp2 = await self._queues[domain].get()
                                         LOGGER.debug(

tests/test_ble_message_routing.py

@@ -0,0 +1,95 @@
+"""Regression tests for VehicleBluetooth._on_message routing and filtering.
+
+Covers VCSEC-vs-INFO domain routing, dropping messages not addressed to us
+(broadcast filtering), and the domain KeyError crash fixed in bluetooth.py:
+``Domain`` has values (DOMAIN_BROADCAST, DOMAIN_AUTHD, ...) with no entry in
+``_queues``, so a message addressed to us but from one of those domains must
+be dropped rather than raising and aborting the reassembly loop mid-buffer.
+"""
+
+from unittest import TestCase
+from unittest.mock import MagicMock
+
+from cryptography.hazmat.primitives.asymmetric import ec
+
+from tesla_fleet_api.tesla.vehicle.bluetooth import VehicleBluetooth
+from tesla_fleet_api.tesla.vehicle.proto.universal_message_pb2 import (
+    Destination,
+    Domain,
+    RoutableMessage,
+)
+
+VIN = "5YJXCAE43LF123456"
+
+
+def _make_vehicle() -> VehicleBluetooth:
+    parent = MagicMock()
+    parent.private_key = ec.generate_private_key(ec.SECP256R1())
+    return VehicleBluetooth(parent, VIN)
+
+
+class OnMessageRoutingTests(TestCase):
+    def test_vcsec_reply_routed_to_vehicle_security_queue(self) -> None:
+        vehicle = _make_vehicle()
+        msg = RoutableMessage(
+            to_destination=Destination(routing_address=vehicle._from_destination),
+            from_destination=Destination(domain=Domain.DOMAIN_VEHICLE_SECURITY),
+        )
+
+        vehicle._on_message(msg)
+
+        self.assertEqual(vehicle._queues[Domain.DOMAIN_VEHICLE_SECURITY].qsize(), 1)
+        self.assertEqual(vehicle._queues[Domain.DOMAIN_INFOTAINMENT].qsize(), 0)
+
+    def test_infotainment_reply_routed_to_infotainment_queue(self) -> None:
+        vehicle = _make_vehicle()
+        msg = RoutableMessage(
+            to_destination=Destination(routing_address=vehicle._from_destination),
+            from_destination=Destination(domain=Domain.DOMAIN_INFOTAINMENT),
+        )
+
+        vehicle._on_message(msg)
+
+        self.assertEqual(vehicle._queues[Domain.DOMAIN_INFOTAINMENT].qsize(), 1)
+        self.assertEqual(vehicle._queues[Domain.DOMAIN_VEHICLE_SECURITY].qsize(), 0)
+
+    def test_message_not_addressed_to_us_is_dropped(self) -> None:
+        vehicle = _make_vehicle()
+        msg = RoutableMessage(
+            to_destination=Destination(routing_address=b"someone-elses-address"),
+            from_destination=Destination(domain=Domain.DOMAIN_INFOTAINMENT),
+        )
+
+        vehicle._on_message(msg)
+
+        self.assertEqual(vehicle._queues[Domain.DOMAIN_INFOTAINMENT].qsize(), 0)
+        self.assertEqual(vehicle._queues[Domain.DOMAIN_VEHICLE_SECURITY].qsize(), 0)
+
+    def test_message_addressed_to_us_from_unhandled_domain_is_dropped_not_raised(
+        self,
+    ) -> None:
+        vehicle = _make_vehicle()
+        msg = RoutableMessage(
+            to_destination=Destination(routing_address=vehicle._from_destination),
+            from_destination=Destination(domain=Domain.DOMAIN_BROADCAST),
+        )
+
+        # Must not raise KeyError - DOMAIN_BROADCAST has no entry in _queues.
+        vehicle._on_message(msg)
+
+        self.assertEqual(vehicle._queues[Domain.DOMAIN_VEHICLE_SECURITY].qsize(), 0)
+        self.assertEqual(vehicle._queues[Domain.DOMAIN_INFOTAINMENT].qsize(), 0)
+
+    def test_message_addressed_to_us_from_authd_domain_is_dropped_not_raised(
+        self,
+    ) -> None:
+        vehicle = _make_vehicle()
+        msg = RoutableMessage(
+            to_destination=Destination(routing_address=vehicle._from_destination),
+            from_destination=Destination(domain=Domain.DOMAIN_AUTHD),
+        )
+
+        vehicle._on_message(msg)
+
+        self.assertEqual(vehicle._queues[Domain.DOMAIN_VEHICLE_SECURITY].qsize(), 0)
+        self.assertEqual(vehicle._queues[Domain.DOMAIN_INFOTAINMENT].qsize(), 0)

tests/test_ble_reassembling_buffer.py

@@ -0,0 +1,116 @@
+"""Regression tests for ``ReassemblingBuffer`` framing and corruption resync.
+
+Exercises the length-prefixed reassembly directly (no VehicleBluetooth, no
+GATT) to lock down: a message split across multiple BLE notification
+chunks, multiple complete messages delivered in a single chunk, and
+resynchronization after a corrupted/oversized packet using the
+``packet_starts`` boundary tracking in ``discard_packet``.
+"""
+
+from unittest import TestCase
+
+from tesla_fleet_api.tesla.vehicle.bluetooth import ReassemblingBuffer, prependLength
+from tesla_fleet_api.tesla.vehicle.proto.universal_message_pb2 import (
+    Destination,
+    Domain,
+    RoutableMessage,
+)
+
+
+def framed(msg: RoutableMessage) -> bytearray:
+    """Length-prefix a RoutableMessage the way the vehicle would frame it."""
+    return prependLength(msg.SerializeToString())
+
+
+class ReassemblingBufferTests(TestCase):
+    def setUp(self) -> None:
+        self.received: list[RoutableMessage] = []
+        self.buffer = ReassemblingBuffer(self.received.append)
+
+    def test_single_message_single_chunk(self) -> None:
+        msg = RoutableMessage(
+            from_destination=Destination(domain=Domain.DOMAIN_INFOTAINMENT)
+        )
+
+        self.buffer.receive_data(framed(msg))
+
+        self.assertEqual(len(self.received), 1)
+        self.assertEqual(
+            self.received[0].from_destination.domain, Domain.DOMAIN_INFOTAINMENT
+        )
+
+    def test_message_split_across_multiple_chunks(self) -> None:
+        msg = RoutableMessage(
+            from_destination=Destination(domain=Domain.DOMAIN_VEHICLE_SECURITY),
+            request_uuid=b"0123456789abcdef",
+        )
+        payload = framed(msg)
+
+        # Simulate GATT MTU fragmentation: deliver a few bytes at a time.
+        chunk_size = 5
+        for i in range(0, len(payload), chunk_size):
+            self.buffer.receive_data(payload[i : i + chunk_size])
+            if i + chunk_size < len(payload):
+                self.assertEqual(
+                    self.received, [], "must not fire callback before full message"
+                )
+
+        self.assertEqual(len(self.received), 1)
+        self.assertEqual(
+            self.received[0].from_destination.domain, Domain.DOMAIN_VEHICLE_SECURITY
+        )
+        self.assertEqual(self.received[0].request_uuid, b"0123456789abcdef")
+
+    def test_multiple_messages_in_one_chunk(self) -> None:
+        first = RoutableMessage(
+            from_destination=Destination(domain=Domain.DOMAIN_VEHICLE_SECURITY)
+        )
+        second = RoutableMessage(
+            from_destination=Destination(domain=Domain.DOMAIN_INFOTAINMENT)
+        )
+
+        self.buffer.receive_data(framed(first) + framed(second))
+
+        self.assertEqual(len(self.received), 2)
+        self.assertEqual(
+            self.received[0].from_destination.domain, Domain.DOMAIN_VEHICLE_SECURITY
+        )
+        self.assertEqual(
+            self.received[1].from_destination.domain, Domain.DOMAIN_INFOTAINMENT
+        )
+
+    def test_corrupt_packet_resyncs_to_next_packet_boundary(self) -> None:
+        good = RoutableMessage(
+            from_destination=Destination(domain=Domain.DOMAIN_INFOTAINMENT)
+        )
+
+        # A plausible-length header (so it's not caught by the oversized
+        # check) whose payload is not a valid RoutableMessage.
+        corrupt = prependLength(b"\xff" * 20)
+
+        # Each call to receive_data is one physical BLE notification, so
+        # this records two distinct packet_starts entries for resync.
+        self.buffer.receive_data(corrupt)
+        self.buffer.receive_data(framed(good))
+
+        self.assertEqual(len(self.received), 1)
+        self.assertEqual(
+            self.received[0].from_destination.domain, Domain.DOMAIN_INFOTAINMENT
+        )
+
+    def test_oversized_length_header_discards_and_resyncs(self) -> None:
+        good = RoutableMessage(
+            from_destination=Destination(domain=Domain.DOMAIN_VEHICLE_SECURITY)
+        )
+
+        # Length header claims a message > 1024 bytes - must be treated as
+        # corrupt rather than stalling forever waiting for more data.
+        oversized_header = bytearray([0xFF, 0xFF]) + b"\x00" * 10
+
+        self.buffer.receive_data(oversized_header)
+        self.buffer.receive_data(framed(good))
+
+        self.assertEqual(len(self.received), 1)
+        self.assertEqual(
+            self.received[0].from_destination.domain, Domain.DOMAIN_VEHICLE_SECURITY
+        )