From 3a6258752e6b8b929a70ef50c702b0e13c3b8974 Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Thu, 23 Jul 2026 20:54:43 +1000 Subject: [PATCH 1/6] fix(ble): route subscription pushes around the command reply queue A vehicleDataSubscription push is addressed to us on the same INFOTAINMENT domain an ordinary command's reply arrives on. _send's pre-send drain of _queues silently discarded any push sitting unconsumed, and _await_response could return a push as an unrelated command's own reply (it satisfies the same HasField(requires) check). Add a per-subscription _StreamSink registry keyed by request_uuid; _on_message peels off correlated pushes before they reach _queues, so neither failure mode can happen. Bounded, drop-oldest per sink since telemetry has no vehicle-side backpressure. --- tesla_fleet_api/tesla/vehicle/bluetooth.py | 58 +++++++- tests/test_ble_stream_sink.py | 150 +++++++++++++++++++++ 2 files changed, 207 insertions(+), 1 deletion(-) create mode 100644 tests/test_ble_stream_sink.py diff --git a/tesla_fleet_api/tesla/vehicle/bluetooth.py b/tesla_fleet_api/tesla/vehicle/bluetooth.py index 074a8e2..6513f8d 100644 --- a/tesla_fleet_api/tesla/vehicle/bluetooth.py +++ b/tesla_fleet_api/tesla/vehicle/bluetooth.py @@ -312,6 +312,36 @@ 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 + + +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. + """ + + def __init__(self, maxsize: int = _STREAM_SINK_MAXSIZE) -> None: + self._queue: asyncio.Queue[RoutableMessage] = asyncio.Queue(maxsize=maxsize) + self.dropped = 0 + + 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 +469,7 @@ class VehicleBluetooth( client: BleakClient | None = None _queues: dict[Domain, asyncio.Queue[RoutableMessage]] _broadcast_watchers: dict[Domain, Callable[[RoutableMessage], None]] + _stream_sinks: dict[bytes, _StreamSink] _ekey: ec.EllipticCurvePublicKey _buffer: ReassemblingBuffer _auth_method = "aes" @@ -512,6 +543,7 @@ def __init__( Domain.DOMAIN_INFOTAINMENT: asyncio.Queue(), } self._broadcast_watchers = {} + self._stream_sinks = {} self._init_broadcast_listeners() self.device = device self._connect_lock = asyncio.Lock() @@ -675,7 +707,15 @@ 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. + """ if msg.to_destination.routing_address != self._from_destination: LOGGER.debug("Ignoring broadcast message (not addressed to us)") @@ -690,6 +730,12 @@ 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: + LOGGER.debug(f"Received subscription push: {msg}") + sink.put(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 +749,16 @@ 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) -> _StreamSink: + """Start routing pushes for ``request_uuid`` into their own sink.""" + sink = _StreamSink() + self._stream_sinks[request_uuid] = sink + return sink + + def _unregister_stream_sink(self, request_uuid: bytes) -> None: + """Stop routing pushes for ``request_uuid``; drops any future pushes.""" + self._stream_sinks.pop(request_uuid, None) + async def _send( self, msg: RoutableMessage, diff --git a/tests/test_ble_stream_sink.py b/tests/test_ble_stream_sink.py new file mode 100644 index 0000000..869821d --- /dev/null +++ b/tests/test_ble_stream_sink.py @@ -0,0 +1,150 @@ +"""Tests for the subscription-push/command-reply queue collision. + +A live ``vehicleDataSubscription`` pushes addressed frames on the same +INFOTAINMENT domain an ordinary command uses for its reply. Before the +``_stream_sinks`` routing fix, ``_send`` would discard any push sitting +unconsumed in ``_queues`` before every send, and ``_await_response`` could +return a push as an unrelated command's own reply. These tests drive the +real (unmocked) ``_send``/``_on_message`` state machine - only the GATT +client is faked - so pushes are injected exactly as ``_on_notify``/the +reassembling buffer would deliver them from a real notification. +""" + +from __future__ import annotations + +import asyncio +from random import randbytes +from typing import Any +from unittest import IsolatedAsyncioTestCase +from unittest.mock import AsyncMock, MagicMock + +from cryptography.hazmat.primitives.asymmetric import ec + +from tesla_fleet_api.tesla.vehicle.bluetooth import VehicleBluetooth +from tesla_protocol.command.universal_message_pb2 import ( + Destination, + Domain, + RoutableMessage, +) + +VIN = "5YJXCAE43LF123456" +DOMAIN = Domain.DOMAIN_INFOTAINMENT + + +def _make_vehicle(**kwargs: Any) -> VehicleBluetooth[Any]: + """A VehicleBluetooth with real send/routing logic but a faked GATT client.""" + parent = MagicMock() + parent.private_key = ec.generate_private_key(ec.SECP256R1()) + vehicle = VehicleBluetooth(parent, VIN, **kwargs) + vehicle.connect_if_needed = AsyncMock() # type: ignore[method-assign] + vehicle.client = MagicMock() + vehicle.client.write_gatt_char = AsyncMock() + return vehicle + + +def _outgoing_msg() -> RoutableMessage: + return RoutableMessage( + to_destination=Destination(domain=DOMAIN), + uuid=randbytes(16), + ) + + +def _addressed_reply(vehicle: VehicleBluetooth[Any]) -> RoutableMessage: + """An addressed reply to an ordinary command.""" + return RoutableMessage( + to_destination=Destination( + domain=DOMAIN, routing_address=vehicle._from_destination + ), + from_destination=Destination(domain=DOMAIN), + protobuf_message_as_bytes=b"command-reply", + ) + + +def _subscription_push( + vehicle: VehicleBluetooth[Any], request_uuid: bytes +) -> RoutableMessage: + """An addressed subscription push, correlated by the subscribe request's uuid.""" + return RoutableMessage( + to_destination=Destination( + domain=DOMAIN, routing_address=vehicle._from_destination + ), + from_destination=Destination(domain=DOMAIN), + protobuf_message_as_bytes=b"push-data", + request_uuid=request_uuid, + ) + + +class StreamSinkQueueCollisionTests(IsolatedAsyncioTestCase): + async def test_already_arrived_push_survives_a_concurrent_send(self) -> None: + vehicle = _make_vehicle() + subscribe_uuid = randbytes(16) + sink = vehicle._register_stream_sink(subscribe_uuid) + + # A push arrives and piles up in the sink before any other command runs. + vehicle._on_message(_subscription_push(vehicle, subscribe_uuid)) + self.assertFalse(sink.empty()) + + async def write_then_ack(*_: Any) -> None: + vehicle._on_message(_addressed_reply(vehicle)) + await asyncio.sleep(0) + + vehicle.client.write_gatt_char = AsyncMock(side_effect=write_then_ack) + + resp = await asyncio.wait_for( + vehicle._send(_outgoing_msg(), "protobuf_message_as_bytes"), timeout=1.0 + ) + + # The command got its own real reply, and the earlier push is still + # sitting in the sink - not silently discarded by _send's pre-send + # drain of _queues. + self.assertEqual(resp.protobuf_message_as_bytes, b"command-reply") + self.assertFalse(sink.empty()) + received = await asyncio.wait_for(sink.get(), timeout=0.1) + self.assertEqual(received.request_uuid, subscribe_uuid) + self.assertEqual(received.protobuf_message_as_bytes, b"push-data") + + async def test_push_arriving_mid_wait_is_not_returned_as_command_reply( + self, + ) -> None: + vehicle = _make_vehicle() + subscribe_uuid = randbytes(16) + sink = vehicle._register_stream_sink(subscribe_uuid) + + async def write_then_push_then_ack(*_: Any) -> None: + # A push lands while the command is still waiting for its reply. + vehicle._on_message(_subscription_push(vehicle, subscribe_uuid)) + await asyncio.sleep(0) + vehicle._on_message(_addressed_reply(vehicle)) + + vehicle.client.write_gatt_char = AsyncMock(side_effect=write_then_push_then_ack) + + resp = await asyncio.wait_for( + vehicle._send(_outgoing_msg(), "protobuf_message_as_bytes"), timeout=1.0 + ) + + # A push carries protobuf_message_as_bytes too, so pre-fix it could + # satisfy _await_response's HasField(requires) check and be returned + # as the command's own reply. It must not be. + self.assertEqual(resp.protobuf_message_as_bytes, b"command-reply") + self.assertFalse(sink.empty()) + + async def test_unregistered_sink_no_longer_receives_pushes(self) -> None: + vehicle = _make_vehicle() + subscribe_uuid = randbytes(16) + sink = vehicle._register_stream_sink(subscribe_uuid) + vehicle._unregister_stream_sink(subscribe_uuid) + + vehicle._on_message(_subscription_push(vehicle, subscribe_uuid)) + + self.assertTrue(sink.empty()) + + async def test_sink_drops_oldest_when_full(self) -> None: + vehicle = _make_vehicle() + subscribe_uuid = randbytes(16) + sink = vehicle._register_stream_sink(subscribe_uuid) + + for _ in range(sink._queue.maxsize + 5): + vehicle._on_message(_subscription_push(vehicle, subscribe_uuid)) + + self.assertEqual(sink.dropped, 5) + self.assertEqual(sink._queue.qsize(), sink._queue.maxsize) From 3423cc7445b4b95cbb3f396a768a1998d18f3abc Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Thu, 23 Jul 2026 20:55:18 +1000 Subject: [PATCH 2/6] docs(agents): document _stream_sinks queue-collision routing --- AGENTS.md | 1 + 1 file changed, 1 insertion(+) diff --git a/AGENTS.md b/AGENTS.md index 23b4e89..7945e6a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -146,6 +146,7 @@ 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`. +- **`_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`, 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. ## Maintaining this file From 489ef87f7f2fe09c75446e1b73087ef40ac5bbef Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Thu, 23 Jul 2026 21:00:13 +1000 Subject: [PATCH 3/6] no-mistakes(review): Make stream sink lifecycle race-safe --- tesla_fleet_api.egg-info/SOURCES.txt | 3 +- tesla_fleet_api/tesla/vehicle/bluetooth.py | 38 +++++++++++++++++----- tests/test_ble_stream_sink.py | 38 +++++++++++++++++++++- 3 files changed, 69 insertions(+), 10 deletions(-) diff --git a/tesla_fleet_api.egg-info/SOURCES.txt b/tesla_fleet_api.egg-info/SOURCES.txt index 73f22d8..fd2c561 100644 --- a/tesla_fleet_api.egg-info/SOURCES.txt +++ b/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 \ No newline at end of file +tests/test_vehicle_image_state.py diff --git a/tesla_fleet_api/tesla/vehicle/bluetooth.py b/tesla_fleet_api/tesla/vehicle/bluetooth.py index 6513f8d..fc88b62 100644 --- a/tesla_fleet_api/tesla/vehicle/bluetooth.py +++ b/tesla_fleet_api/tesla/vehicle/bluetooth.py @@ -469,7 +469,7 @@ class VehicleBluetooth( client: BleakClient | None = None _queues: dict[Domain, asyncio.Queue[RoutableMessage]] _broadcast_watchers: dict[Domain, Callable[[RoutableMessage], None]] - _stream_sinks: dict[bytes, _StreamSink] + _stream_sinks: dict[bytes, _StreamSink | None] _ekey: ec.EllipticCurvePublicKey _buffer: ReassemblingBuffer _auth_method = "aes" @@ -730,11 +730,15 @@ 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: - LOGGER.debug(f"Received subscription push: {msg}") - sink.put(msg) - return + if msg.request_uuid in self._stream_sinks: + sink = self._stream_sinks[msg.request_uuid] + if sink is None: + LOGGER.debug(f"Dropping push for retired subscription: {msg}") + return + if msg.HasField("protobuf_message_as_bytes"): + LOGGER.debug(f"Received subscription push: {msg}") + sink.put(msg) + return queue = self._queues.get(msg.from_destination.domain) if queue is None: @@ -756,8 +760,26 @@ def _register_stream_sink(self, request_uuid: bytes) -> _StreamSink: return sink def _unregister_stream_sink(self, request_uuid: bytes) -> None: - """Stop routing pushes for ``request_uuid``; drops any future pushes.""" - self._stream_sinks.pop(request_uuid, None) + """Retire ``request_uuid`` so delayed pushes are dropped.""" + self._stream_sinks[request_uuid] = None + + 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.""" + sink = self._register_stream_sink(msg.uuid) + 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, diff --git a/tests/test_ble_stream_sink.py b/tests/test_ble_stream_sink.py index 869821d..6acdce1 100644 --- a/tests/test_ble_stream_sink.py +++ b/tests/test_ble_stream_sink.py @@ -60,6 +60,18 @@ def _addressed_reply(vehicle: VehicleBluetooth[Any]) -> RoutableMessage: ) +def _subscription_ack( + vehicle: VehicleBluetooth[Any], request_uuid: bytes +) -> RoutableMessage: + return RoutableMessage( + to_destination=Destination( + domain=DOMAIN, routing_address=vehicle._from_destination + ), + from_destination=Destination(domain=DOMAIN), + request_uuid=request_uuid, + ) + + def _subscription_push( vehicle: VehicleBluetooth[Any], request_uuid: bytes ) -> RoutableMessage: @@ -128,7 +140,30 @@ async def write_then_push_then_ack(*_: Any) -> None: self.assertEqual(resp.protobuf_message_as_bytes, b"command-reply") self.assertFalse(sink.empty()) - async def test_unregistered_sink_no_longer_receives_pushes(self) -> None: + async def test_subscription_ack_completes_atomic_registration(self) -> None: + vehicle = _make_vehicle() + outgoing = _outgoing_msg() + + async def write_then_push_then_ack(*_: Any) -> None: + vehicle._on_message(_subscription_push(vehicle, outgoing.uuid)) + vehicle._on_message(_subscription_ack(vehicle, outgoing.uuid)) + + vehicle.client.write_gatt_char = AsyncMock(side_effect=write_then_push_then_ack) + + response, sink = await asyncio.wait_for( + vehicle._send_with_stream_sink( + outgoing, "protobuf_message_as_bytes", timeout=1.0 + ), + timeout=1.0, + ) + + self.assertEqual(response.request_uuid, outgoing.uuid) + self.assertEqual( + (await asyncio.wait_for(sink.get(), timeout=0.1)).protobuf_message_as_bytes, + b"push-data", + ) + + async def test_unregistered_sink_drops_late_pushes(self) -> None: vehicle = _make_vehicle() subscribe_uuid = randbytes(16) sink = vehicle._register_stream_sink(subscribe_uuid) @@ -137,6 +172,7 @@ async def test_unregistered_sink_no_longer_receives_pushes(self) -> None: vehicle._on_message(_subscription_push(vehicle, subscribe_uuid)) self.assertTrue(sink.empty()) + self.assertTrue(vehicle._queues[DOMAIN].empty()) async def test_sink_drops_oldest_when_full(self) -> None: vehicle = _make_vehicle() From 7a9451efa000190d6916ea86c297d694780757b6 Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Thu, 23 Jul 2026 21:02:26 +1000 Subject: [PATCH 4/6] no-mistakes(review): Bound retired stream routing tombstones --- tesla_fleet_api/tesla/vehicle/bluetooth.py | 24 +++++++++++++++------- tests/test_ble_stream_sink.py | 14 +++++++++++++ 2 files changed, 31 insertions(+), 7 deletions(-) diff --git a/tesla_fleet_api/tesla/vehicle/bluetooth.py b/tesla_fleet_api/tesla/vehicle/bluetooth.py index fc88b62..fdd3175 100644 --- a/tesla_fleet_api/tesla/vehicle/bluetooth.py +++ b/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 @@ -315,6 +316,7 @@ def _plan_auto_conditioning(action: VehicleAction) -> VerifyPlan | None: # 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: @@ -469,7 +471,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 | None] + _stream_sinks: dict[bytes, _StreamSink] + _retired_streams: deque[bytes] _ekey: ec.EllipticCurvePublicKey _buffer: ReassemblingBuffer _auth_method = "aes" @@ -544,6 +547,7 @@ def __init__( } 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() @@ -730,15 +734,15 @@ def _on_message(self, msg: RoutableMessage) -> None: self._dispatch_status_listeners(status) return - if msg.request_uuid in self._stream_sinks: - sink = self._stream_sinks[msg.request_uuid] - if sink is None: - LOGGER.debug(f"Dropping push for retired subscription: {msg}") - return + sink = self._stream_sinks.get(msg.request_uuid) + if sink is not None: if msg.HasField("protobuf_message_as_bytes"): LOGGER.debug(f"Received subscription push: {msg}") sink.put(msg) return + 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: @@ -756,12 +760,18 @@ def _on_message(self, msg: RoutableMessage) -> None: def _register_stream_sink(self, request_uuid: bytes) -> _StreamSink: """Start routing pushes for ``request_uuid`` into their own sink.""" sink = _StreamSink() + 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[request_uuid] = None + 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, diff --git a/tests/test_ble_stream_sink.py b/tests/test_ble_stream_sink.py index 6acdce1..96cac5f 100644 --- a/tests/test_ble_stream_sink.py +++ b/tests/test_ble_stream_sink.py @@ -174,6 +174,20 @@ async def test_unregistered_sink_drops_late_pushes(self) -> None: self.assertTrue(sink.empty()) self.assertTrue(vehicle._queues[DOMAIN].empty()) + async def test_retired_stream_routes_are_bounded(self) -> None: + vehicle = _make_vehicle() + retired = [randbytes(16) for _ in range(vehicle._retired_streams.maxlen + 5)] + + for request_uuid in retired: + vehicle._register_stream_sink(request_uuid) + vehicle._unregister_stream_sink(request_uuid) + + self.assertEqual( + len(vehicle._retired_streams), vehicle._retired_streams.maxlen + ) + self.assertNotIn(retired[0], vehicle._retired_streams) + self.assertIn(retired[-1], vehicle._retired_streams) + async def test_sink_drops_oldest_when_full(self) -> None: vehicle = _make_vehicle() subscribe_uuid = randbytes(16) From ed5fbb72c87f7da7a7b58f468ed64b012e24889b Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Thu, 23 Jul 2026 21:05:39 +1000 Subject: [PATCH 5/6] no-mistakes(document): Refresh streaming documentation and formatting --- AGENTS.md | 2 +- tests/test_ble_stream_sink.py | 4 +--- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 7945e6a..f55e5fc 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -147,7 +147,7 @@ Keep the `tesla-protocol` floor at `>=0.5.0`; earlier releases have generated `. - **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`. - **`_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`, 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. +- **`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. diff --git a/tests/test_ble_stream_sink.py b/tests/test_ble_stream_sink.py index 96cac5f..53aa882 100644 --- a/tests/test_ble_stream_sink.py +++ b/tests/test_ble_stream_sink.py @@ -182,9 +182,7 @@ async def test_retired_stream_routes_are_bounded(self) -> None: vehicle._register_stream_sink(request_uuid) vehicle._unregister_stream_sink(request_uuid) - self.assertEqual( - len(vehicle._retired_streams), vehicle._retired_streams.maxlen - ) + self.assertEqual(len(vehicle._retired_streams), vehicle._retired_streams.maxlen) self.assertNotIn(retired[0], vehicle._retired_streams) self.assertIn(retired[-1], vehicle._retired_streams) From 652a82e4c50726410141a6a0da4cd1688196f85c Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Thu, 23 Jul 2026 21:21:30 +1000 Subject: [PATCH 6/6] fix(ble): let a subscribe request's own ack pass through its sink _send_with_stream_sink registered its sink armed from the start, so the subscribe request's ack - which carries the same request_uuid as every later push - got diverted into the sink instead of reaching _send's reply queue, and the subscribe call timed out waiting for a reply the vehicle had already sent. Sinks created via _send_with_stream_sink now start unarmed; _on_message lets exactly the first matching frame fall through to the ordinary reply queue and arms the sink immediately after, so every frame from then on is treated as a push. --- tesla_fleet_api/tesla/vehicle/bluetooth.py | 52 +++++++++++++++---- tests/test_ble_stream_sink.py | 59 +++++++++++++++++++--- 2 files changed, 93 insertions(+), 18 deletions(-) diff --git a/tesla_fleet_api/tesla/vehicle/bluetooth.py b/tesla_fleet_api/tesla/vehicle/bluetooth.py index fdd3175..1439e15 100644 --- a/tesla_fleet_api/tesla/vehicle/bluetooth.py +++ b/tesla_fleet_api/tesla/vehicle/bluetooth.py @@ -326,11 +326,21 @@ class _StreamSink: 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) -> None: + 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(): @@ -719,6 +729,11 @@ def _on_message(self, msg: RoutableMessage) -> None: 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: @@ -736,10 +751,14 @@ def _on_message(self, msg: RoutableMessage) -> None: sink = self._stream_sinks.get(msg.request_uuid) if sink is not None: - if msg.HasField("protobuf_message_as_bytes"): - LOGGER.debug(f"Received subscription push: {msg}") - sink.put(msg) - return + 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 @@ -757,9 +776,17 @@ 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) -> _StreamSink: - """Start routing pushes for ``request_uuid`` into their own sink.""" - sink = _StreamSink() + 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: @@ -780,8 +807,13 @@ async def _send_with_stream_sink( *, timeout: float | None = None, ) -> tuple[RoutableMessage, _StreamSink]: - """Atomically register a stream route and send its subscription request.""" - sink = self._register_stream_sink(msg.uuid) + """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 diff --git a/tests/test_ble_stream_sink.py b/tests/test_ble_stream_sink.py index 53aa882..0cc4970 100644 --- a/tests/test_ble_stream_sink.py +++ b/tests/test_ble_stream_sink.py @@ -63,11 +63,15 @@ def _addressed_reply(vehicle: VehicleBluetooth[Any]) -> RoutableMessage: def _subscription_ack( vehicle: VehicleBluetooth[Any], request_uuid: bytes ) -> RoutableMessage: + """The subscribe request's own ack - a real one carries + ``protobuf_message_as_bytes`` too (an infotainment actuation's + ``actionStatus`` rides in that field), the same as every later push.""" return RoutableMessage( to_destination=Destination( domain=DOMAIN, routing_address=vehicle._from_destination ), from_destination=Destination(domain=DOMAIN), + protobuf_message_as_bytes=b"subscribe-ack", request_uuid=request_uuid, ) @@ -140,28 +144,67 @@ async def write_then_push_then_ack(*_: Any) -> None: self.assertEqual(resp.protobuf_message_as_bytes, b"command-reply") self.assertFalse(sink.empty()) - async def test_subscription_ack_completes_atomic_registration(self) -> None: + async def test_subscribe_ack_is_not_swallowed_by_its_own_sink(self) -> None: + # The subscribe request's own ack carries the same request_uuid as + # every later push (both have protobuf_message_as_bytes set), so a + # sink armed immediately on registration would divert that first ack + # into itself instead of letting `_send` consume it - and + # `_send_with_stream_sink` would then time out waiting for a reply + # that had already arrived. The sink must start unarmed and only + # divert pushes that arrive *after* the ack. vehicle = _make_vehicle() outgoing = _outgoing_msg() - async def write_then_push_then_ack(*_: Any) -> None: - vehicle._on_message(_subscription_push(vehicle, outgoing.uuid)) + async def write_then_ack(*_: Any) -> None: vehicle._on_message(_subscription_ack(vehicle, outgoing.uuid)) + await asyncio.sleep(0) - vehicle.client.write_gatt_char = AsyncMock(side_effect=write_then_push_then_ack) + vehicle.client.write_gatt_char = AsyncMock(side_effect=write_then_ack) + # A short timeout proves this resolves from the injected ack rather + # than from a fallback timeout path. response, sink = await asyncio.wait_for( vehicle._send_with_stream_sink( - outgoing, "protobuf_message_as_bytes", timeout=1.0 + outgoing, "protobuf_message_as_bytes", timeout=0.5 ), timeout=1.0, ) + self.assertEqual(response.protobuf_message_as_bytes, b"subscribe-ack") self.assertEqual(response.request_uuid, outgoing.uuid) - self.assertEqual( - (await asyncio.wait_for(sink.get(), timeout=0.1)).protobuf_message_as_bytes, - b"push-data", + + # The subscription is now live: a push carrying the same + # request_uuid must reach the sink, not be treated as another ack. + vehicle._on_message(_subscription_push(vehicle, outgoing.uuid)) + pushed = await asyncio.wait_for(sink.get(), timeout=0.1) + self.assertEqual(pushed.protobuf_message_as_bytes, b"push-data") + + async def test_push_arriving_before_ack_still_arms_the_sink(self) -> None: + # Not the expected wire ordering (the vehicle sends the ack before + # any push), but the sink must not get stuck unarmed forever if a + # frame beats the ack: whichever frame arrives first satisfies + # _send_with_stream_sink's wait, and everything after it is treated + # as a push. + vehicle = _make_vehicle() + outgoing = _outgoing_msg() + + async def write_then_first_frame(*_: Any) -> None: + vehicle._on_message(_subscription_push(vehicle, outgoing.uuid)) + await asyncio.sleep(0) + + vehicle.client.write_gatt_char = AsyncMock(side_effect=write_then_first_frame) + + response, sink = await asyncio.wait_for( + vehicle._send_with_stream_sink( + outgoing, "protobuf_message_as_bytes", timeout=0.5 + ), + timeout=1.0, ) + self.assertEqual(response.protobuf_message_as_bytes, b"push-data") + + vehicle._on_message(_subscription_push(vehicle, outgoing.uuid)) + pushed = await asyncio.wait_for(sink.get(), timeout=0.1) + self.assertEqual(pushed.protobuf_message_as_bytes, b"push-data") async def test_unregistered_sink_drops_late_pushes(self) -> None: vehicle = _make_vehicle()