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

[Bre77]

Intent

Fix a confirmed BLE streaming bug: VehicleBluetooth._send empties the DOMAIN_INFOTAINMENT reply queue before every send, so any command sent while a vehicleDataSubscription is live silently discards pushed subscription frames sitting in that queue - and, separately, _await_response could wrongly return a push as an unrelated command's own reply, since a push also satisfies the HasField('protobuf_message_as_bytes') check. This is PR A ('streaming-queue-collision-fix') of a larger, already-completed streaming-api-design plan; the public streaming API (context-manager/async-iterator, decrypt, renewal loop) is deliberately out of scope for this PR and is filed as separate follow-up work. Fix: added a _StreamSink (bounded asyncio.Queue, drop-oldest policy, dropped counter) and a _stream_sinks registry on VehicleBluetooth keyed by the subscribe request's request_uuid. _on_message now checks this registry before routing an addressed frame into _queues - a match is diverted into its own sink and never enters _queues, so _send's pre-send drain can't discard it and _await_response can't return it as a different command's reply. Added _register_stream_sink/_unregister_stream_sink as the only entry points into the registry (no public subscription API yet - that's later PRs). Live-BLE re-verification against a real vehicle was explicitly deferred by the captain (hardware testing deprioritized); this lands the code fix plus a unit test only. Added tests/test_ble_stream_sink.py: drives the real (unmocked) _send/_on_message state machine with only the GATT client faked, proving (1) a push that arrived before a concurrent command survives that command's send/drain, (2) a push arriving mid-wait is not returned as the command's own reply, (3) unregistering a sink stops routing to it, (4) the sink drops oldest entries once full. Verified these tests fail with AttributeError against the pre-fix code (the registry/methods didn't exist). Also added one AGENTS.md bullet documenting the _stream_sinks routing scheme for future streaming.py work. Full local validation before opening this run: ruff check, ruff format --check, pyright tesla_fleet_api (0 errors), and the full pytest suite (446 passed) all green.

What Changed

• Route BLE subscription pushes into bounded, per-request sinks so command queue draining and reply matching cannot discard or misidentify streamed data.
• Add race-safe subscription registration and bounded tombstones that drop delayed pushes after a stream is retired.
• Add stream-routing regression coverage and refresh internal routing documentation and the package source manifest.

Risk Assessment

✅ Low: The updated change is well-bounded, resolves the subscription startup and teardown routing races, and now bounds retired-stream state without introducing a material regression.

Testing

Author-reported lint, formatting, type-check, and prior full-suite validation were complemented by independent focused BLE tests, a reviewer-visible mocked-GATT state-machine demonstration, and the complete 448-test suite; everything passed and the worktree remained clean. Live vehicle testing was intentionally deferred per the supplied intent.

Evidence: BLE stream/command collision demonstration

{ "command_received": "actual-command-reply", "subscription_pushes_preserved_in_order": [ "push-before-command", "push-during-command" ], "shared_command_queue_empty": true, "overflow_policy": { "dropped": 5, "oldest_retained": "snapshot-5" }, "late_push_after_unregister_routed": false }

{
  "command_received": "actual-command-reply",
  "subscription_pushes_preserved_in_order": [
    "push-before-command",
    "push-during-command"
  ],
  "shared_command_queue_empty": true,
  "overflow_policy": {
    "dropped": 5,
    "oldest_retained": "snapshot-5"
  },
  "late_push_after_unregister_routed": false
}

Pipeline

Updates from git push no-mistakes

✅ **intent** - passed

✅ No issues found.

✅ **Rebase** - passed

✅ No issues found.

🔧 **Review** - 2 issues found → auto-fixed (2) ✅

• 🚨 tesla_fleet_api/tesla/vehicle/bluetooth.py:733 - Routing every addressed frame whose request_uuid is registered creates an impossible registration race: registering before sending the subscription diverts that subscription's own ACK/error away from _await_response, while registering afterward allows early pushes to enter _queues. Provide an atomic subscribe-send path that distinguishes the initial command response from subsequent pushes.
• ⚠️ tesla_fleet_api/tesla/vehicle/bluetooth.py:760 - Unregistering removes all knowledge of the stream UUID, so a delayed push immediately falls back into the shared command-reply queue and can again be discarded or mistaken for another command's reply. Preserve a terminal/tombstoned route long enough to drop late frames, or otherwise ensure cancellation and routing teardown are synchronized; the current docstring's claim that future pushes are dropped is incorrect.

🔧 Fix: Make stream sink lifecycle race-safe
1 warning still open:

• ⚠️ tesla_fleet_api/tesla/vehicle/bluetooth.py:764 - Retired subscription UUIDs are retained as None forever. Once the planned renewal loop repeatedly creates fresh request UUIDs, this registry grows without bound for the lifetime of a VehicleBluetooth instance. Use a bounded/expiring tombstone collection so late pushes remain isolated without leaking one entry per subscription renewal.

🔧 Fix: Bound retired stream routing tombstones
✅ Re-checked - no issues remain.

✅ **Test** - passed

✅ No issues found.

• Inspected git diff edc2faf2490f52d27c99a53da650f254b369f0bf..4e6c19bcd39383925a2e958c5f823a5c1e6cd3e6 against the supplied acceptance criteria
• uv run pytest -q tests/test_ble_stream_sink.py tests/test_ble_send_transport.py tests/test_ble_broadcast_confirmation.py tests/test_ble_write_timeout_router.py
• uv run python /tmp/no-mistakes-evidence/01KY79ZRXP72WENBBZD5SFVQZA/stream_sink_demo.py
• uv run pytest -q
• git status --short
• git diff --check edc2faf2490f52d27c99a53da650f254b369f0bf..4e6c19bcd39383925a2e958c5f823a5c1e6cd3e6

✅ **Document** - passed

✅ No issues found.

✅ **Lint** - passed

✅ No issues found.

✅ **Push** - passed

✅ No issues found.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 21990d09df

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[chatgpt-codex-connector]

Preserve the subscription response before diverting pushes

When establishing a vehicle-data subscription whose initial response is a normal Response.vehicleDataSubscriptionResponse payload, that response has the same request_uuid and protobuf_message_as_bytes as later pushes, so this branch sends it into the stream sink instead of the domain reply queue. _send_with_stream_sink() then keeps waiting for the subscribe response until it times out, even though the car replied; reserve the first matching response for _send (or distinguish the decoded response from subsequent stream pushes) before routing ongoing pushes to the sink.

Useful? React with 👍 / 👎.

[Bre77]

Codex's P2 finding (bluetooth.py:741) was correct: _send_with_stream_sink registered its sink armed from the start, and the subscribe request's own ack shares its request_uuid with every later push (both carry protobuf_message_as_bytes), so that first ack got diverted into the sink instead of reaching _send - _send_with_stream_sink then timed out waiting for a reply the vehicle had already sent.

Fixed in 652a82e: _StreamSink gained an armed flag. _send_with_stream_sink now registers its sink unarmed, so _on_message lets exactly the first matching frame (the ack) fall through to the normal reply queue and arms the sink right after - every frame after that is a genuine push and gets diverted as before.

Added two regression tests (tests/test_ble_stream_sink.py): one with the realistic wire ordering (ack first, then a push) asserting the subscribe call resolves under a short timeout instead of hanging, and one covering the sink still arming correctly if a frame beats the ack. Both fail with BluetoothTimeout against the pre-fix code and pass after. Also fixed the _subscription_ack test helper, which previously omitted protobuf_message_as_bytes and so accidentally masked this exact bug in the original regression test.

Full suite green locally (456 passed, up from 446).