fix: raise on 401/403 instead of retrying, and fix README's undocumented async-with usage

AGENTS.md

@@ -20,6 +20,8 @@ This file is the project's committed home for project-intrinsic agent knowledge:
 - `TeslemetryStream.async_add_listener(..., internal=True)` marks a listener as bookkeeping-only (the vehicle's config-sync listener is the only current user). Its `schedule_refresh` gate on the `asyncio.create_task()` call is unconditionally false for `internal=True`, regardless of loop state - this is what makes eager, construction-time registration of the config listener safe outside a running event loop, not deferred timing. Both the "first listener starts the task" and "last listener removed auto-closes" checks also count only non-internal (public) listeners for the same underlying reason: an internal listener alone must not itself start the task, and one surviving an auto-close must not block a later public listener's own zero-to-one transition from restarting it. Any stream stand-in passed to `TeslemetryStreamVehicle` (test doubles included) must implement `async_add_listener(callback, filters, internal=False)` and `async_add_connection_listener(callback)`.
 - `add_field` gates its no-op skip on `TeslemetryStreamVehicle._populated`, not on connection/topic state: an unpopulated vehicle awaits `_ensure_populated()` (a single-flight `get_config()` REST fetch - concurrent callers, e.g. a batch of `listen_*` calls at HA integration setup, join one GET instead of each starting their own) before deciding; a populated one trusts `fields` outright with no network call. `_populated` is set by a successful `get_config()` (200 or 404 - both are an authoritative answer; 404 also clears `fields`, since no config existing is itself the authoritative state, not a fetch to ignore) and by every `_on_config_event` push, and cleared by an `_on_connection_event` disconnect notification (registered via `async_add_connection_listener` at construction, alongside the config-sync listener) - a disconnect leaves the record possibly stale until the next connection's config snapshot arrives, so a field-config call landing in that reconnect window re-fetches instead of trusting pre-disconnect data. A failed populating fetch (`aiohttp.ClientError`/timeout, or a non-200/404 status via `raise_for_status()`) is not authoritative like a 404: `_ensure_populated()` catches it, logs, and leaves the vehicle unpopulated rather than letting it propagate - every `listen_*` method reaches `add_field()` through `_enable_field()`'s fire-and-forget `asyncio.create_task()`, where an uncaught exception would just silently abandon the field request instead of sending the PATCH.
 - `tests/test_config_events.py` covers the record merge and nested-entry validation; `tests/test_config_listener_lifecycle.py` covers construction-time registration (including outside a running loop), the auto-close exclusion, and the populated/unpopulated no-op-skip gating; `tests/test_stream_lifecycle.py` covers the internal-listener-survives-close restart case and a vehicle discovered mid-dispatch of its own config event (misses that event, stays unpopulated, and self-corrects via the lazy fetch on its first field-config call); `tests/test_reconnect_config_window.py` covers the reconnect-window race (a field-config call between `connect()` and that connection's config snapshot re-fetches rather than trusting the stale pre-disconnect record).
+- `TeslemetryStream` has no `__aenter__`/`__aexit__` - do not reintroduce `async with TeslemetryStream(...)` in the README or examples. Connection lifecycle is entirely listener-driven: `async_add_listener` connects on the first (public) listener and disconnects on the last one removed; `connect()`/`close()`/`listen()` exist for callers who want to manage the connection themselves instead.
+- `__anext__` treats a `aiohttp.ClientResponseError` with `status` 401 or 403 as terminal, not transient: it sets `active = False` and raises `TeslemetryStreamAuthenticationError` (chaining the original error) rather than retrying, since a rejected token can never succeed on retry. Every other `aiohttp.ClientError` (including other response statuses) keeps the pre-existing backoff-and-reconnect behavior. `tests/test_auth_failure.py` covers both the 401/403 surfacing and that a genuine transient `ClientError` still retries and reconnects.
 - `TeslemetryStream` owns exactly one `_listen_task`: `async_add_listener`'s zero-to-one transition only creates it when absent/done, and `listen()` itself checks `asyncio.current_task()` against `self._listen_task` - a second concurrent `listen()` call joins the owner via `await existing_task` instead of racing it for the connection. `connect()` serializes the actual GET behind `self._connect_lock` and re-checks `self.active` after acquiring both the lock and the response, discarding a response that arrived after a stop/supersede rather than publishing it. Internal reconnect paths (EOF, `ClientError`, unexpected exceptions in `__anext__`) call `_close_response()`, which only clears the response and notifies connection listeners - they must not call `close()`, which additionally flips `active=False` and cancels the owned task, i.e. a real stop. `listen()`'s `finally` calls `_close_response()` unconditionally, so task cancellation (however triggered) still releases the connection. `listen()` dispatches over a *sorted* snapshot of `_listeners.values()` (never the live dict) with internal listeners ordered first, regardless of registration order: a callback that adds a listener mid-dispatch (e.g. `get_vehicle()` for an uncached VIN) must not raise `RuntimeError: dictionary changed size during iteration` and kill the loop, and a public callback must not get a chance to mutate the event in place before an internal (bookkeeping) listener has cached from it. `_update_connection_listeners()` has the same hazard for `_connection_listeners` (a connection listener calling `get_vehicle()` registers that vehicle's own connection listener mid-dispatch) and is fixed the same way, over a plain `list(...)` snapshot - no ordering requirement there, since connection listeners don't share a mutable event object. `tests/test_stream_lifecycle.py` covers the add/remove/re-add race, duplicate `listen()` calls, cancellation while blocked reading content, close-during-connect, close-during-backoff, a listener mutating `_listeners` mid-dispatch, internal-before-public dispatch order, and a connection listener mutating `_connection_listeners` mid-dispatch.
 
 ## Maintaining this file

CLAUDE.md

@@ -0,0 +1,2 @@
+<!-- Points Claude at AGENTS.md via import; edit AGENTS.md, not this file. -->
+@AGENTS.md

README.md

@@ -25,30 +25,30 @@ The TeslemetryStream class requires:
 
 The full list of possible values are provided in `TelemetryFields` and `TelemetryAlerts`
 
-To connect, either use `async with` on the instance, call `connect()`, or register an callback with `async_add_listener`, which will connect when added and disconnect when removed.
+The simplest way to connect is to register a callback with `async_add_listener`: it connects automatically when the first listener is added, and disconnects when the last one is removed - no explicit `connect()`/`close()` call needed.
 
-Using `connect()` or `listen()` will require you to close the session manually using `close()`.
+If you want to manage the connection yourself instead, call `connect()` and drive the stream with `listen()`; in that case you are responsible for calling `close()` when done.
 
 ## Example
 The following example puts the listening loop in the background, then stopping after 20 seconds.
 ```
 async def main():
     async with aiohttp.ClientSession() as session:
-        async with TeslemetryStream(
+        stream = TeslemetryStream(
             access_token="<token>",
             vin="<vin>", # for single vehicles
             server="na.teslemetry.com" # or "eu.teslemetry.com"
             session=session,
-        ) as stream:
+        )
 
-            def callback(event):
-                print(event["data"])
+        def callback(event):
+            print(event["data"])
 
-            remove = stream.async_add_listener(callback)
+        remove = stream.async_add_listener(callback)
 
-            print("Running")
-            await asyncio.sleep(60)
-            remove()
+        print("Running")
+        await asyncio.sleep(60)
+        remove()
 ```
 
 ## Using Typed Listen Methods
@@ -58,27 +58,27 @@ The library provides typed listen methods for various telemetry signals. These m
 ```python
 async def main():
     async with aiohttp.ClientSession() as session:
-        async with TeslemetryStream(
+        stream = TeslemetryStream(
             access_token="<token>",
             vin="<vin>", # for single vehicles
             session=session,
-        ) as stream:
+        )
 
-            vehicle = stream.get_vehicle("<vin>")
+        vehicle = stream.get_vehicle("<vin>")
 
-            def battery_level_callback(battery_level):
-                print(f"Battery Level: {battery_level}")
+        def battery_level_callback(battery_level):
+            print(f"Battery Level: {battery_level}")
 
-            def vehicle_speed_callback(vehicle_speed):
-                print(f"Vehicle Speed: {vehicle_speed}")
+        def vehicle_speed_callback(vehicle_speed):
+            print(f"Vehicle Speed: {vehicle_speed}")
 
-            remove_battery_level_listener = vehicle.listen_BatteryLevel(battery_level_callback)
-            remove_vehicle_speed_listener = vehicle.listen_VehicleSpeed(vehicle_speed_callback)
+        remove_battery_level_listener = vehicle.listen_BatteryLevel(battery_level_callback)
+        remove_vehicle_speed_listener = vehicle.listen_VehicleSpeed(vehicle_speed_callback)
 
-            print("Running")
-            await asyncio.sleep(60)
-            remove_battery_level_listener()
-            remove_vehicle_speed_listener()
+        print("Running")
+        await asyncio.sleep(60)
+        remove_battery_level_listener()
+        remove_vehicle_speed_listener()
 ```
 
 ## Writing Your Own Listener with Multiple Signals

teslemetry_stream/__init__.py

@@ -8,6 +8,7 @@
 )
 from .energysite import TeslemetryStreamEnergySite
 from .exception import (
+    TeslemetryStreamAuthenticationError,
     TeslemetryStreamConnectionError,
     TeslemetryStreamEnded,
     TeslemetryStreamError,
@@ -24,6 +25,7 @@
     "Signal",
     "SseTopic",
     "TeslemetryStream",
+    "TeslemetryStreamAuthenticationError",
     "TeslemetryStreamConnectionError",
     "TeslemetryStreamEnded",
     "TeslemetryStreamEnergySite",

teslemetry_stream/exception.py

@@ -23,3 +23,9 @@ class TeslemetryStreamEnded(TeslemetryStreamError):
   """Teslemetry Stream Connection Error"""
 
   message = "The stream was ended by the server."
+
+
+class TeslemetryStreamAuthenticationError(TeslemetryStreamError):
+  """Teslemetry Stream Authentication Error"""
+
+  message = "The access token was rejected (401/403) and will not be retried."

teslemetry_stream/stream.py

@@ -11,7 +11,7 @@
 
 from .const import CreditsEvent
 from .energysite import TeslemetryStreamEnergySite
-from .exception import TeslemetryStreamEnded
+from .exception import TeslemetryStreamAuthenticationError, TeslemetryStreamEnded
 from .vehicle import TeslemetryStreamVehicle
 
 LOGGER = logging.getLogger(__package__)
@@ -325,6 +325,8 @@ async def __anext__(self) -> dict[str, Any]:
         :return: Next event as a dictionary.
         :raises StopAsyncIteration: If the stream is stopped.
         :raises TeslemetryStreamEnded: If the stream is ended by the server.
+        :raises TeslemetryStreamAuthenticationError: If the access token is
+            rejected with a 401 or 403.
         """
         while self.active:
             try:
@@ -353,6 +355,14 @@ async def __anext__(self) -> dict[str, Any]:
                 LOGGER.warning("Stream ended by server")
                 self._close_response()
             except aiohttp.ClientError as error:
+                if isinstance(error, aiohttp.ClientResponseError) and error.status in (401, 403):
+                    # A rejected token is a definitive answer, not a transient
+                    # blip - retrying it can never succeed, and doing so masks
+                    # a bad credential as an indefinitely quiet stream.
+                    LOGGER.error("Authentication failed, not retrying: %s", repr(error))
+                    self.active = False
+                    self._close_response()
+                    raise TeslemetryStreamAuthenticationError() from error
                 LOGGER.warning("Client error: %s", repr(error))
                 self._close_response()
                 delay = min(2**self.retries, 600)

tests/test_auth_failure.py

@@ -0,0 +1,194 @@
+"""Regression tests for __anext__'s auth-failure handling: a 401/403
+aiohttp.ClientResponseError is a subtype of aiohttp.ClientError, so without
+special-casing it, an invalid token is swallowed and retried exactly like a
+transient network blip - the caller sees no events and no error. These tests
+cover the fix (auth failure surfaces as TeslemetryStreamAuthenticationError
+and is not retried) and its safety rail (a genuine transient ClientError
+still retries and reconnects, unchanged).
+"""
+from __future__ import annotations
+
+import asyncio
+import contextlib
+from typing import Any
+
+import aiohttp
+
+from teslemetry_stream.exception import TeslemetryStreamAuthenticationError
+from teslemetry_stream.stream import TeslemetryStream
+
+REQUEST_INFO = aiohttp.RequestInfo(
+    url="https://fake.teslemetry.com/sse",
+    method="GET",
+    headers={},  # type: ignore[arg-type]
+    real_url="https://fake.teslemetry.com/sse",  # type: ignore[arg-type]
+)
+
+
+class FakeContent:
+    """Async-iterable response body that blocks until failed or cancelled."""
+
+    def __init__(self) -> None:
+        self._blocker: asyncio.Future[None] = asyncio.get_running_loop().create_future()
+
+    def __aiter__(self) -> FakeContent:
+        return self
+
+    async def __anext__(self) -> bytes:
+        await self._blocker
+        raise AssertionError("unreachable - blocker only resolves via an exception")
+
+    def fail(self, exc: BaseException) -> None:
+        if not self._blocker.done():
+            self._blocker.set_exception(exc)
+
+
+class FakeResponse:
+    """Minimal stand-in for the aiohttp response `connect()` awaits."""
+
+    def __init__(self) -> None:
+        self.url = "https://fake.teslemetry.com/sse"
+        self.status = 200
+        self.content = FakeContent()
+        self.closed = False
+
+    def close(self) -> None:
+        self.closed = True
+
+
+class FakeSession:
+    """Captures every `get()` call; a queue of `get_results` (each either an
+    exception to raise or a FakeResponse to return) drives each call in
+    order, mirroring how a real 401 raises straight out of the connect GET
+    when `raise_for_status=True`."""
+
+    def __init__(self, get_results: list[Any]) -> None:
+        self.calls = 0
+        self._get_results = list(get_results)
+        self.responses: list[FakeResponse] = []
+
+    async def get(self, url: str, **kwargs: Any) -> FakeResponse:
+        self.calls += 1
+        result = self._get_results.pop(0)
+        if isinstance(result, BaseException):
+            raise result
+        self.responses.append(result)
+        return result
+
+
+def make_stream(session: FakeSession) -> TeslemetryStream:
+    return TeslemetryStream(
+        session=session,  # type: ignore[arg-type]
+        access_token="bad-token",
+        server="api.teslemetry.com",
+        manual=True,
+    )
+
+
+def check(label: str, ok: bool, detail: str = "") -> bool:
+    print(f"{label:<72} {'PASS' if ok else 'FAIL'}{'  ' + detail if detail else ''}")
+    return ok
+
+
+async def drain_cancelled(task: asyncio.Task[Any]) -> None:
+    with contextlib.suppress(asyncio.CancelledError):
+        await task
+
+
+async def test_401_surfaces_and_does_not_retry(results: list[bool]) -> None:
+    error = aiohttp.ClientResponseError(
+        request_info=REQUEST_INFO, history=(), status=401, message="Unauthorized"
+    )
+    session = FakeSession([error])
+    stream = make_stream(session)
+
+    task = asyncio.create_task(stream.listen())
+
+    raised: Exception | None = None
+    try:
+        await task
+    except TeslemetryStreamAuthenticationError as exc:
+        raised = exc
+
+    results.append(
+        check("a 401 surfaces as TeslemetryStreamAuthenticationError", raised is not None)
+    )
+    results.append(
+        check(
+            "the original 401 is chained as the cause",
+            raised is not None and raised.__cause__ is error,
+        )
+    )
+    results.append(
+        check("no retry is attempted after a 401", session.calls == 1, f"got {session.calls}")
+    )
+    results.append(check("the stream stops rather than looping forever", not stream.active))
+
+
+async def test_403_surfaces_and_does_not_retry(results: list[bool]) -> None:
+    error = aiohttp.ClientResponseError(
+        request_info=REQUEST_INFO, history=(), status=403, message="Forbidden"
+    )
+    session = FakeSession([error])
+    stream = make_stream(session)
+
+    task = asyncio.create_task(stream.listen())
+
+    raised: Exception | None = None
+    try:
+        await task
+    except TeslemetryStreamAuthenticationError as exc:
+        raised = exc
+
+    results.append(
+        check("a 403 surfaces as TeslemetryStreamAuthenticationError", raised is not None)
+    )
+    results.append(
+        check("no retry is attempted after a 403", session.calls == 1, f"got {session.calls}")
+    )
+
+
+async def test_transient_client_error_still_retries_and_reconnects(results: list[bool]) -> None:
+    """A genuine transient failure (connection reset mid-stream, not a 401/403
+    response) must keep retrying and reconnecting exactly as before - the
+    auth-failure special-case must not touch this path."""
+    session = FakeSession([FakeResponse(), FakeResponse()])
+    stream = make_stream(session)
+
+    task = asyncio.create_task(stream.listen())
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+    results.append(check("initial connect happened", session.calls == 1, f"got {session.calls}"))
+
+    session.responses[0].content.fail(aiohttp.ClientError("boom"))
+    # retries starts at 0, so the first backoff delay is 2**0 == 1 second.
+    await asyncio.sleep(1.2)
+
+    results.append(check("the stream is still active after a transient error", stream.active))
+    results.append(
+        check(
+            "the stream reconnected instead of surfacing an error",
+            session.calls == 2,
+            f"got {session.calls}",
+        )
+    )
+    results.append(check("the listen task is still running", not task.done()))
+
+    stream.close()
+    await drain_cancelled(task)
+
+
+async def main() -> None:
+    results: list[bool] = []
+    await test_401_surfaces_and_does_not_retry(results)
+    await test_403_surfaces_and_does_not_retry(results)
+    await test_transient_client_error_still_retries_and_reconnects(results)
+
+    print("-" * 72)
+    print("ALL PASS" if all(results) else "FAILURES PRESENT")
+    if not all(results):
+        raise SystemExit(1)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())