fix(vehicle): isolate BLE subscription pushes from command replies

AGENTS.md

@@ -146,7 +146,8 @@ Keep the `tesla-protocol` floor at `>=0.5.0`; earlier releases have generated `.
 - **`_log_request_result` (`fleet.py`) must tolerate any JSON-legal REST body, not just dicts**: it runs after the HTTP request already succeeded, so it's a logging convenience only - a non-dict body (`null`, a list, a bare scalar; live case: Teslemetry's `list_authorized_clients` returning `null`) must never raise there. It guards with `isinstance(data, dict)` before calling `.get()`, logging `result=success` and returning for anything else. Regression tests in `tests/test_command_logging.py` (`test_null_json_body_returns_none_without_raising` etc.) cover null/list/scalar bodies.
 - **Typed accessor pattern for undocumented raw-dict responses**: `TeslemetryEnergySite.find_authorized_clients()` (`teslemetry/energysite.py`) is the library's first frozen-dataclass typed wrapper over a raw `dict[str, Any]`/`None`/`list` REST response, added so API-parsing logic (envelope unwrap, field lookup, shape validation, enum typing) lives in the library instead of each consumer (e.g. Home Assistant's config flow) reimplementing it. `_authorized_clients_list()` does one precise, non-recursive envelope unwrap (`{"response": {"authorized_clients": [...]}}` or `{"response": {"clients": [...]}}`, or a bare list with no envelope) rather than a multi-key/depth-first search - the `clients` key variant was added after a live capture (Tesla Release 953) showed the endpoint's real key differs from the originally-documented `authorized_clients`, confirmed against a populated 5-entry sample. `AuthorizedClient` models only the two fields a pairing flow actually reads (`public_key`, `state`), each accepting only the specific key-name variant pairs empirically evidenced, not speculative extras. Two rules any future typed accessor over an undocumented response shape must keep, demonstrated here even though `AuthorizedClientState` (`const.py`) has no falsy member: (1) field lookup must check key presence (`key in payload`), never `payload.get(key) or default` - a legal falsy value is not "missing", and a present-but-unrecognized enum value (`_normalize_state()`) is returned raw rather than coerced to `None`; (2) a `None` body and an unrecognized response shape (not a dict/list, or an envelope that unwraps to neither accepted list key) are malformed data, not "zero clients" - Tesla's endpoint intermittently returns HTTP 200 with a null body, so `_authorized_clients_list()` raises `InvalidResponse` (`exceptions.py`) for both rather than collapsing them to `[]`; only a genuinely empty list under either accepted key parses to `AuthorizedClients.clients == []` without raising. Tesla has not published an OpenAPI schema for pairing endpoints, so `const.py`'s enums are the schema of record; widen `AuthorizedClient`'s modeled fields only against a further live sample, not speculatively. The raw `list_authorized_clients()` method is kept alongside as the untyped escape hatch and still returns a null body unchanged rather than raising. Tests: `tests/test_teslemetry_authorized_clients.py`.
 - **`networking_status`'s `ipv4_config` fields are raw big-endian uint32 ints, not strings**: the wire format applies to all of `ipv4_config.address`/`subnet_mask`/`gateway`, but `TeslemetryEnergySite.find_gateway_address()` (`teslemetry/energysite.py`, second typed accessor after `find_authorized_clients()`, same rules) decodes only `address` - confirmed against a live Powerwall 3 capture where `3232235914` decodes network-byte-order (`struct.pack(">I", ...)`) to `192.168.1.138`, not little-endian. It considers only `eth`/`wifi` (never `gsm` - cellular isn't a LAN path), preferring whichever has `active_route` set and a decodable address, else falling back to the first of the two (in that order) with any decodable address; `0`/`0xFFFFFFFF` are treated as undecodable so an unconfigured interface never shadows a real one with `0.0.0.0`. A `{"response": null}` envelope raises `InvalidResponse` (the endpoint's known intermittent malformed mode), while a well-formed response with no usable interface returns `None` (not a raise). Tests: `tests/test_teslemetry_gateway_address.py`.
-- **`VehicleAction`/`GetVehicleData` proto coverage is locked by test, not just by convention**: `tests/test_proto_coverage_lock.py` walks both descriptors and fails if any field has no wrapper (`commands.py`) or reader (`bluetooth.py`) and isn't on one of its two small, reasoned allowlists - keep that test in sync with any future `tesla-protocol` bump rather than special-casing new fields elsewhere. The only fields deliberately left unwrapped today are the 7-field push-style subscription/streaming family (`createStreamSession`/`streamMessage`/`vehicleDataSubscription`/`vehicleDataAck`/`vitalsSubscription`/`vitalsAck`/`cancelVehicleDataSubscription`, needs a persistent INFOTAINMENT-domain broadcast dispatcher like `broadcast.py`'s VCSEC listeners) and `getVehicleImageState` (needs chunked binary-transfer paging) - both are separate, unscoped design work, not oversights. CarServer's `GetVehicleState` sub-state is exposed as `legacy_vehicle_state()` (`bluetooth.py`), matching the `VehicleData.legacy_vehicle_state` reply field name, specifically to avoid confusion with the pre-existing `vehicle_state()` (VCSEC `VehicleStatus`, a different message/domain). `set_rate_tariff`/`add_managed_charging_site` (`commands.py`) take `tesla_protocol` message types directly for their deeply-nested arguments rather than a parallel flattened dataclass API. `pii_key_request`/`pseudonym_sync_request`/`tesla_auth_response`/`setup_cloud_profile_with_local_profile_uuid`/`get_local_profiles_for_vault_uuid` are wrapped with no known third-party consumer use case, purely for full-proto-coverage completeness.
+- **`_stream_sinks` peels subscription pushes off the command-reply queue before routing**: a live `vehicleDataSubscription`'s pushes arrive addressed to us on the same domain queue (`_queues`) an ordinary command's reply uses, correlated by the subscribe request's own `request_uuid`. `_on_message` (`bluetooth.py`) checks `self._stream_sinks.get(msg.request_uuid)` before touching `_queues` - a match routes into that subscription's own bounded, drop-oldest `_StreamSink` instead, so `_send`'s pre-send drain can never discard a push and `_await_response` can never return one as an unrelated command's reply. `_register_stream_sink`/`_unregister_stream_sink` are the only entry points into the registry today; there is no public subscription API yet (`VehicleDataSubscription`/`createStreamSession`/`cancelVehicleDataSubscription` remain unwrapped, see the proto-coverage entry below) - this is dispatch-layer plumbing for that future work. Tests: `tests/test_ble_stream_sink.py`.
+- **`VehicleAction`/`GetVehicleData` proto coverage is locked by test, not just by convention**: `tests/test_proto_coverage_lock.py` walks both descriptors and fails if any field has no wrapper (`commands.py`) or reader (`bluetooth.py`) and isn't on one of its two small, reasoned allowlists - keep that test in sync with any future `tesla-protocol` bump rather than special-casing new fields elsewhere. The only fields deliberately left unwrapped today are the 7-field push-style subscription/streaming family (`createStreamSession`/`streamMessage`/`vehicleDataSubscription`/`vehicleDataAck`/`vitalsSubscription`/`vitalsAck`/`cancelVehicleDataSubscription`, which still need a public lifecycle/iterator API atop the private `_stream_sinks` routing described above) and `getVehicleImageState` (needs chunked binary-transfer paging) - both are separate, unscoped design work, not oversights. CarServer's `GetVehicleState` sub-state is exposed as `legacy_vehicle_state()` (`bluetooth.py`), matching the `VehicleData.legacy_vehicle_state` reply field name, specifically to avoid confusion with the pre-existing `vehicle_state()` (VCSEC `VehicleStatus`, a different message/domain). `set_rate_tariff`/`add_managed_charging_site` (`commands.py`) take `tesla_protocol` message types directly for their deeply-nested arguments rather than a parallel flattened dataclass API. `pii_key_request`/`pseudonym_sync_request`/`tesla_auth_response`/`setup_cloud_profile_with_local_profile_uuid`/`get_local_profiles_for_vault_uuid` are wrapped with no known third-party consumer use case, purely for full-proto-coverage completeness.
 ## Maintaining this file
 
 Keep this file for knowledge useful to almost every future agent session in this project.

tesla_fleet_api.egg-info/SOURCES.txt

@@ -75,6 +75,7 @@ tests/test_ble_optimistic_and_best_effort.py
 tests/test_ble_pair.py
 tests/test_ble_reassembling_buffer.py
 tests/test_ble_send_transport.py
+tests/test_ble_stream_sink.py
 tests/test_ble_ui_unit_preference_commands.py
 tests/test_ble_unconfirmed_command.py
 tests/test_ble_write_timeout_router.py
@@ -94,4 +95,4 @@ tests/test_tesla_private_key.py
 tests/test_teslemetry_authorized_clients.py
 tests/test_teslemetry_gateway_address.py
 tests/test_tessie_vehicle_params.py
-tests/test_vehicle_image_state.py
+tests/test_vehicle_image_state.py

tesla_fleet_api/tesla/vehicle/bluetooth.py

@@ -5,6 +5,7 @@
 import struct
 import time
 import warnings
+from collections import deque
 from random import randbytes
 from typing import TYPE_CHECKING, Any, Callable, Generic, TypeVar
 
@@ -312,6 +313,47 @@ def _plan_auto_conditioning(action: VehicleAction) -> VerifyPlan | None:
     "hvacAutoAction": _plan_auto_conditioning,
 }
 
+# A subscription push stream has no vehicle-side backpressure, so an unbounded
+# sink behind a stalled consumer would grow without limit.
+_STREAM_SINK_MAXSIZE = 32
+_STREAM_TOMBSTONE_MAXSIZE = 32
+
+
+class _StreamSink:
+    """Non-draining queue for one subscription's addressed pushes.
+
+    Bounded and drop-oldest, since the newest snapshot supersedes older ones
+    for telemetry. Kept entirely separate from ``_queues`` so ``_send``'s
+    pre-send drain and the one-shot ``_await_response`` consumer can never
+    see or discard a subscription push.
+
+    ``armed`` gates whether ``_on_message`` diverts a matching frame into this
+    sink at all. The subscribe request's own ack shares its ``request_uuid``
+    with every later push, so a sink registered before that ack arrives must
+    start unarmed - letting exactly that first frame fall through to the
+    ordinary reply queue for ``_send`` to consume - and only arm once the ack
+    has been claimed.
+    """
+
+    def __init__(
+        self, maxsize: int = _STREAM_SINK_MAXSIZE, *, armed: bool = True
+    ) -> None:
+        self._queue: asyncio.Queue[RoutableMessage] = asyncio.Queue(maxsize=maxsize)
+        self.dropped = 0
+        self.armed = armed
+
+    def put(self, msg: RoutableMessage) -> None:
+        if self._queue.full():
+            self._queue.get_nowait()
+            self.dropped += 1
+        self._queue.put_nowait(msg)
+
+    async def get(self) -> RoutableMessage:
+        return await self._queue.get()
+
+    def empty(self) -> bool:
+        return self._queue.empty()
+
 
 class VehicleBluetooth(
     BroadcastListeners, Commands[BluetoothParentT], Generic[BluetoothParentT]
@@ -439,6 +481,8 @@ class VehicleBluetooth(
     client: BleakClient | None = None
     _queues: dict[Domain, asyncio.Queue[RoutableMessage]]
     _broadcast_watchers: dict[Domain, Callable[[RoutableMessage], None]]
+    _stream_sinks: dict[bytes, _StreamSink]
+    _retired_streams: deque[bytes]
     _ekey: ec.EllipticCurvePublicKey
     _buffer: ReassemblingBuffer
     _auth_method = "aes"
@@ -512,6 +556,8 @@ def __init__(
             Domain.DOMAIN_INFOTAINMENT: asyncio.Queue(),
         }
         self._broadcast_watchers = {}
+        self._stream_sinks = {}
+        self._retired_streams = deque(maxlen=_STREAM_TOMBSTONE_MAXSIZE)
         self._init_broadcast_listeners()
         self.device = device
         self._connect_lock = asyncio.Lock()
@@ -675,7 +721,20 @@ def _on_notify(self, sender: BleakGATTCharacteristic, data: bytearray) -> None:
         self._buffer.receive_data(data)
 
     def _on_message(self, msg: RoutableMessage) -> None:
-        """Route addressed BLE replies into the per-domain response queue."""
+        """Route addressed BLE replies into the per-domain response queue.
+
+        A subscription push is addressed to us and carries its subscribe
+        request's own ``request_uuid``, so it is peeled off into that
+        subscription's own sink before it can reach ``_queues`` - keeping it
+        out of reach of both ``_send``'s pre-send drain and the one-shot
+        ``_await_response`` consumer, which would otherwise discard it or
+        return it as an unrelated command's reply.
+
+        The subscribe request's own ack shares that same ``request_uuid``
+        with later pushes, so an unarmed sink lets exactly one matching frame
+        fall through here to ``_queues`` - the ack ``_send_with_stream_sink``
+        is waiting on - before arming itself for every push after it.
+        """
 
         if msg.to_destination.routing_address != self._from_destination:
             LOGGER.debug("Ignoring broadcast message (not addressed to us)")
@@ -690,6 +749,20 @@ def _on_message(self, msg: RoutableMessage) -> None:
                     self._dispatch_status_listeners(status)
             return
 
+        sink = self._stream_sinks.get(msg.request_uuid)
+        if sink is not None:
+            if sink.armed:
+                if msg.HasField("protobuf_message_as_bytes"):
+                    LOGGER.debug(f"Received subscription push: {msg}")
+                    sink.put(msg)
+                    return
+            else:
+                LOGGER.debug(f"Received subscribe ack, arming sink: {msg}")
+                sink.armed = True
+        elif msg.request_uuid in self._retired_streams:
+            LOGGER.debug(f"Dropping push for retired subscription: {msg}")
+            return
+
         queue = self._queues.get(msg.from_destination.domain)
         if queue is None:
             # Domain enum has values (e.g. DOMAIN_BROADCAST, DOMAIN_AUTHD) with
@@ -703,6 +776,53 @@ def _on_message(self, msg: RoutableMessage) -> None:
         LOGGER.debug(f"Received response: {msg}")
         queue.put_nowait(msg)
 
+    def _register_stream_sink(
+        self, request_uuid: bytes, *, armed: bool = True
+    ) -> _StreamSink:
+        """Start routing pushes for ``request_uuid`` into their own sink.
+
+        ``armed=False`` is for a sink registered ahead of its own subscribe
+        request: the first matching frame (that request's ack) is left to
+        fall through to ``_queues`` instead of being diverted, and
+        ``_on_message`` arms the sink once it has passed through.
+        """
+        sink = _StreamSink(armed=armed)
+        try:
+            self._retired_streams.remove(request_uuid)
+        except ValueError:
+            pass
+        self._stream_sinks[request_uuid] = sink
+        return sink
+
+    def _unregister_stream_sink(self, request_uuid: bytes) -> None:
+        """Retire ``request_uuid`` so delayed pushes are dropped."""
+        self._stream_sinks.pop(request_uuid, None)
+        if request_uuid not in self._retired_streams:
+            self._retired_streams.append(request_uuid)
+
+    async def _send_with_stream_sink(
+        self,
+        msg: RoutableMessage,
+        requires: str,
+        *,
+        timeout: float | None = None,
+    ) -> tuple[RoutableMessage, _StreamSink]:
+        """Atomically register a stream route and send its subscription request.
+
+        The registered sink starts unarmed so this request's own ack - which
+        carries the same ``request_uuid`` as every later push - is left for
+        ``_send`` to consume as an ordinary reply rather than being diverted.
+        """
+        sink = self._register_stream_sink(msg.uuid, armed=False)
+        try:
+            response = await self._send(
+                msg, requires, expects_data=False, timeout=timeout
+            )
+        except BaseException:
+            self._unregister_stream_sink(msg.uuid)
+            raise
+        return response, sink
+
     async def _send(
         self,
         msg: RoutableMessage,