feat(vehicle): add VehicleRouter for primary/fallback command dispatch

[Bre77]

Intent

The developer wanted a new "routing"/failover Vehicle class added to the tesla_fleet_api library that composes an ordered primary and fallback pair of existing Vehicle instances and dynamically dispatches each method call: routing to the primary when a health check passes, falling back otherwise, and transparently falling through to the fallback for methods the primary lacks (raising AttributeError only when neither has it). The concrete use case was pairing a VehicleBluetooth primary with a TeslemetryVehicle cloud fallback, so commands go over BLE when reachable and route to the cloud when not. Requirements included a configurable health check accepting a bool, sync callable, or async callable (defaulting to a feature-detected primary check that treats BLE "can connect" as healthy without hard-coding Bluetooth assumptions), correct async awaiting throughout, pyright-strict type hints matching the sibling classes' generic style, and keeping the design generic enough to later wrap Energy site classes. The developer required pytest tests using mocks/fakes (no real BLE or network) covering all dispatch and health-check cases, proper exports alongside sibling vehicle classes, and the change committed on the branch with the suite passing. During validation via the no-mistakes pipeline, the captain approved upgrading from probe-only failover to true per-command exception failover, which the developer then drove through the gate (including correctly catching Tesla faults that inherit from BaseException rather than Exception).

What Changed

• Add VehicleRouter (tesla/vehicle/router.py), a composition wrapper that pairs a primary and fallback Vehicle instance and dispatches each method call to the primary when a health check passes, with automatic per-command failover to the fallback on any TeslaFleetError/exception (including Tesla faults that inherit from BaseException); methods absent on the primary fall through to the fallback, raising AttributeError only when neither has them. The health check accepts a bool, sync callable, or async callable, defaulting to a feature-detected primary probe.
• Export VehicleRouter from tesla/vehicle/__init__.py alongside the sibling vehicle classes (no factory is added to the Vehicles collections).
• Document the router in the README and architecture docs, including caveats around potential double-execution of non-idempotent commands on replay and that dunder methods (e.g. __aenter__/__aexit__) are not proxied; add coverage in tests/test_vehicle_router.py.

Risk Assessment

✅ Low: The change is a well-bounded, additive composition wrapper with comprehensive tests covering all dispatch/health/failover paths, correct handling of BaseException-derived Tesla faults, and previously-raised concerns already documented per an explicit prior decision.

Testing

Ran the existing 15-case pytest suite for VehicleRouter (all pass, covering primary/fallback dispatch, bool/sync/async health checks, per-command failover including Tesla faults that subclass BaseException, and CancelledError propagation), then built and ran an end-to-end demo that models the concrete intended scenario — a BLE primary paired with a cloud fallback — and captured a CLI transcript showing each command visibly routed to [BLE] or [CLOUD] across all nine dispatch/health/failover paths, plus a smoke check confirming the public export. This is a library change with no UI surface, so the reviewer-visible evidence is a CLI transcript of the routing decisions rather than a screenshot; the intent is satisfied end-to-end.

Evidence: End-to-end router routing transcript (BLE primary + cloud fallback)

1. Primary (BLE) reachable -> [BLE] flashed lights 2. Primary (BLE) unreachable -> [CLOUD] flashed lights (per-command failover) 3. Mid-command BluetoothTimeout (BaseException) -> [CLOUD] flashed lights 4. Health via primary.can_connect: False -> [CLOUD] honk; True -> [BLE] honk 5. Async health False -> [CLOUD] flashed lights 6. Cloud-only wake() -> falls through to [CLOUD] woke 7. router.vin -> prefers PRIMARY value 8. Unknown attr -> AttributeError (routes to neither primary nor fallback) 9. Both transports fail -> ConnectionError propagated

====================================================================
1. Primary (BLE) reachable, no health gate -> routes to PRIMARY
====================================================================
flash_lights() -> [BLE] flashed lights on 5YJ3E1EA7KF000001

====================================================================
2. Primary (BLE) unreachable -> per-command failover to FALLBACK
====================================================================
flash_lights() -> [CLOUD] flashed lights on 5YJ3E1EA7KF000001

====================================================================
3. Mid-command Tesla fault (BaseException subclass) -> FALLBACK
====================================================================
flash_lights() -> [CLOUD] flashed lights on 5YJ3E1EA7KF000001

====================================================================
4. Explicit health check via primary.can_connect (feature-detected)
====================================================================
can_connect() = False (unhealthy)
honk_horn() -> [CLOUD] honked horn on 5YJ3E1EA7KF000001 (primary skipped entirely)
can_connect() = True (healthy)
honk_horn() -> [BLE] honked horn on 5YJ3E1EA7KF000001

====================================================================
5. Async health check returning False -> FALLBACK
====================================================================
await is_healthy() = False
flash_lights() -> [CLOUD] flashed lights on 5YJ3E1EA7KF000001

====================================================================
6. Method only on FALLBACK (cloud wake) -> falls through, no health gate
====================================================================
wake() -> [CLOUD] woke 5YJ3E1EA7KF000001

====================================================================
7. Non-callable attribute (vin) -> prefers PRIMARY
====================================================================
router.vin = 5YJ3E1EA7KF000001

====================================================================
8. Attribute on NEITHER -> AttributeError
====================================================================
AttributeError: 'VehicleRouter' routes to neither primary 'FakeBluetoothVehicle' nor fallback 'FakeCloudVehicle' for 'nonexistent_method'

====================================================================
9. Both transports fail -> original error propagates
====================================================================
ConnectionError propagated: cloud API unavailable

All routing scenarios demonstrated.

Evidence: Demo script

"""End-to-end demonstration of VehicleRouter's health-gated primary/fallback
dispatch, using fakes that stand in for a VehicleBluetooth (BLE) primary and a
TeslemetryVehicle cloud fallback. No real BLE hardware or network is used.

Run: uv run python /tmp/.../demo_router.py
"""
import asyncio

from tesla_fleet_api.exceptions import BluetoothTimeout
from tesla_fleet_api.tesla.vehicle.router import VehicleRouter


class FakeBluetoothVehicle:
    """Stands in for VehicleBluetooth: BLE transport, may be unreachable."""

    def __init__(self, vin, *, reachable=True, mid_command_fault=None):
        self.vin = vin
        self.reachable = reachable
        self.mid_command_fault = mid_command_fault

    def can_connect(self):
        # Feature-detected health signal: "can the BLE link be established?"
        return self.reachable

    async def flash_lights(self):
        if self.mid_command_fault is not None:
            raise self.mid_command_fault
        if not self.reachable:
            raise ConnectionError("BLE device not in range")
        return f"[BLE] flashed lights on {self.vin}"

    async def honk_horn(self):
        if not self.reachable:
            raise ConnectionError("BLE device not in range")
        return f"[BLE] honked horn on {self.vin}"


class FakeCloudVehicle:
    """Stands in for TeslemetryVehicle: cloud transport, always reachable here."""

    def __init__(self, vin):
        self.vin = vin

    async def flash_lights(self):
        return f"[CLOUD] flashed lights on {self.vin}"

    async def honk_horn(self):
        return f"[CLOUD] honked horn on {self.vin}"

    async def wake(self):
        # Cloud-only capability, not present on the BLE primary.
        return f"[CLOUD] woke {self.vin}"


def line(title):
    print("\n" + "=" * 68)
    print(title)
    print("=" * 68)


async def main():
    VIN = "5YJ3E1EA7KF000001"

    line("1. Primary (BLE) reachable, no health gate -> routes to PRIMARY")
    router = VehicleRouter(
        FakeBluetoothVehicle(VIN, reachable=True), FakeCloudVehicle(VIN)
    )
    print("flash_lights() ->", await router.flash_lights())

    line("2. Primary (BLE) unreachable -> per-command failover to FALLBACK")
    router = VehicleRouter(
        FakeBluetoothVehicle(VIN, reachable=False), FakeCloudVehicle(VIN)
    )
    print("flash_lights() ->", await router.flash_lights())

    line("3. Mid-command Tesla fault (BaseException subclass) -> FALLBACK")
    # BluetoothTimeout inherits from TeslaFleetError/BaseException, not Exception.
    router = VehicleRouter(
        FakeBluetoothVehicle(VIN, reachable=True, mid_command_fault=BluetoothTimeout()),
        FakeCloudVehicle(VIN),
    )
    print("flash_lights() ->", await router.flash_lights())

    line("4. Explicit health check via primary.can_connect (feature-detected)")
    primary = FakeBluetoothVehicle(VIN, reachable=False)
    router = VehicleRouter(primary, FakeCloudVehicle(VIN), health=primary.can_connect)
    print("can_connect() =", primary.can_connect(), "(unhealthy)")
    print("honk_horn() ->", await router.honk_horn(), "(primary skipped entirely)")
    primary.reachable = True
    print("can_connect() =", primary.can_connect(), "(healthy)")
    print("honk_horn() ->", await router.honk_horn())

    line("5. Async health check returning False -> FALLBACK")

    async def unhealthy():
        return False

    router = VehicleRouter(
        FakeBluetoothVehicle(VIN, reachable=True), FakeCloudVehicle(VIN),
        health=unhealthy,
    )
    print("await is_healthy() =", await router.is_healthy())
    print("flash_lights() ->", await router.flash_lights())

    line("6. Method only on FALLBACK (cloud wake) -> falls through, no health gate")
    router = VehicleRouter(
        FakeBluetoothVehicle(VIN, reachable=True), FakeCloudVehicle(VIN)
    )
    print("wake() ->", await router.wake())

    line("7. Non-callable attribute (vin) -> prefers PRIMARY")
    print("router.vin =", router.vin)

    line("8. Attribute on NEITHER -> AttributeError")
    try:
        router.nonexistent_method
    except AttributeError as e:
        print("AttributeError:", e)

    line("9. Both transports fail -> original error propagates")
    router = VehicleRouter(
        FakeBluetoothVehicle(VIN, reachable=False),
        FakeCloudVehicle(VIN),
    )
    # Make the cloud fallback fail too.
    router.fallback.flash_lights = _boom
    try:
        await router.flash_lights()
    except ConnectionError as e:
        print("ConnectionError propagated:", e)

    print("\nAll routing scenarios demonstrated.")


async def _boom():
    raise ConnectionError("cloud API unavailable")


if __name__ == "__main__":
    asyncio.run(main())

Pipeline

Updates from git push no-mistakes

✅ **intent** - passed

✅ No issues found.

✅ **Rebase** - passed

✅ No issues found.

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

• ⚠️ tesla_fleet_api/tesla/vehicle/router.py:107 - Per-command failover replays the same call on the fallback whenever the primary raises, including mid-command transport errors (documented as 'a write/notify failure or a disconnect'). For non-idempotent vehicle commands (honk_horn, actuate_trunk, door_unlock, charge_start, etc.) the primary may have already partially applied the command over BLE before the transport error, so the fallback replay double-executes it. The docstring calls out that mid-command errors trigger failover but does not warn about the double-application consequence. Confirm this is acceptable for the intended command set, or restrict blanket replay to connection-establishment failures.
• ℹ️ tesla_fleet_api/tesla/vehicle/router.py:115 - getattr proxying does not cover special/dunder methods, which Python looks up on the type rather than the instance. The documented use case pairs a VehicleBluetooth primary, which is an async context manager (aenter/aexit manage the BLE connection). async with VehicleRouter(...) will not proxy to the primary and won't manage the connection lifecycle. Commands still work because BLE _send() calls connect_if_needed() internally, but explicit disconnect/cleanup via aexit is lost and users must reach through router.primary. Worth a note in the README/docstring.

🔧 Fix: Document double-execution and dunder-proxy caveats in VehicleRouter
✅ Re-checked - no issues remain.

✅ **Test** - passed

✅ No issues found.

• uv run pytest tests/test_vehicle_router.py -v — all 15 dispatch/health-check/failover tests pass (includes TeslaFleetError/BaseException failover and CancelledError propagation)
• Wrote and ran /tmp/no-mistakes-evidence/01KWGVVJ5RT9367VPRME4Q0CQG/demo_router.py — end-to-end demo with fakes standing in for a VehicleBluetooth primary and TeslemetryVehicle cloud fallback, covering: primary success, per-command failover on unreachable BLE, mid-command Tesla fault (BaseException) failover, feature-detected can_connect health gate, async health check, fallback-only method fall-through, non-callable vin preferring primary, AttributeError on neither, and both-fail error propagation
• uv run python -c 'from tesla_fleet_api.tesla.vehicle import VehicleRouter, VehicleBluetooth, VehicleFleet' — confirms VehicleRouter is exported alongside sibling vehicle classes

✅ **Document** - passed

✅ No issues found.

✅ **Lint** - passed

✅ No issues found.

✅ **Push** - passed

✅ No issues found.