fix: consume config-update SSE events into the vehicle config record

AGENTS.md

@@ -16,6 +16,7 @@ This file is the project's committed home for project-intrinsic agent knowledge:
 - `site_info` events do not carry `tariff_content`/`tariff_content_v2`; the V2 tariff is its own `tariff_content_v2` event/listener (`listen_TariffContentV2`), same envelope shape as `site_info`, with a `None` body meaning an explicit server-side removal rather than "not received yet". Both share the same silence-means-no-change contract - freshness lives in REST, never in event cadence. There is deliberately no library helper recombining `site_info` and `tariff_content_v2` into one document - that would only ever cover the V2 tariff (legacy V1 `tariff_content` has no SSE topic and stays REST-only by design); a consumer wanting both tariffs together should use the REST site_info endpoint.
 - Releases (tag `v*.*.*`) go through `.github/workflows/release.yml` directly - it's the sole top-level workflow, triggered on the tag push: `lint` + the full `test` python-version matrix (mirrors `ci.yml`) must pass on the exact release SHA before `build` (single Python, build+twine) runs, and only then do the `pypi` environment's protection rules (and its trusted-publishing OIDC) allow `publish-to-pypi`. It must stay a top-level workflow, not a `workflow_call` reusable one - PyPI's trusted publisher is configured for the `release.yml` + `pypi` environment identity, and a reusable-workflow caller signs PEP 740 attestations under the caller's identity instead, which that publisher check rejects. The `pypi` GitHub environment itself (required reviewers, deployment branches) is admin-configured outside this repo's files. `jobs.<id>.environment.name`/`.url` cannot reference the `env` context (only `github`, `inputs`, `vars`, `needs`, `secrets`, `strategy`, `matrix` resolve there) - referencing `env.*` there is a workflow-file parse error that fails the whole file at startup, before trigger filtering, so it fails every push (not just tags) with zero jobs and no logs. Use literal values or `vars.*` instead. `release.yml` also carries `workflow_dispatch` so a tag whose run failed before publish can be re-run manually without tag surgery.
 - `TeslemetryStream(topics=...)` is an optional exact SSE wire-event allowlist sent as the connection's `topics` query param; `SseTopic` in `const.py` is the closed set the server recognizes, and `SSE_VEHICLE_TOPICS`/`SSE_ENERGY_TOPICS`/`SSE_ALL_TOPICS` are client-side presets - flat per-product-kind lists of exact wire names, deliberately not further split by whether a topic happens to have a connect-time snapshot server-side; that's upstream server behavior, not something this library encodes. Omitting `topics` (`None`) is legacy-all forever - every applicable event delivered unfiltered. An explicitly empty iterable is rejected with `ValueError` at construction time rather than silently falling back to legacy-all - "no topics" must not mean "all topics", mirroring the server's own 400 on an empty `topics` value. A bare `str`/`SseTopic` is accepted as a single topic rather than iterated character-by-character - `topics` type-checks `str | Iterable[str] | None` precisely because a lone string also satisfies `Iterable[str]`, the classic footgun. `tests/test_sse_topics.py` covers the tariff listener, its null-removal signal, the `topics` param's URL construction, the empty-iterable rejection, and the bare-string/bare-`SseTopic` case.
+- `TeslemetryStreamVehicle` keeps `fields`/`preferTyped` fresh against server-side changes (another client, the console, a Teslemetry migration), not just this client's own history: `_ensure_config_listener()` lazily registers an internal listener (`_on_config_event`, filtered on `{Key.VIN, Key.CONFIG: None}`) on the `config` SSE topic, shaped `{vin, config: {fields, prefer_typed}}` like the REST `get_config` body. A well-typed `fields`/`prefer_typed` piece replaces the corresponding record field; a missing piece is left untouched, and a malformed piece (wrong type) is logged and skipped without touching the other piece or the pending `_config`. Registration is deferred to the first `add_field`/`prefer_typed`/`update_config` call (not `__init__`) because `TeslemetryStream.async_add_listener()` calls `asyncio.create_task()` for a stream's first-ever listener, which needs a running loop - eager construction-time registration broke the previously-synchronous `TeslemetryStream(vin=...)`/vehicle construction. `async_add_listener(..., internal=True)` marks it so it's excluded from the "last listener removed" auto-close check in `TeslemetryStream` - otherwise a permanently-registered internal listener would keep `_listeners` non-empty forever and the SSE connection would never close once every real listener unsubscribed. Because of this, any stream stand-in passed to `TeslemetryStreamVehicle` (test doubles included) must implement `async_add_listener(callback, filters, internal=False)`. `tests/test_config_events.py` covers the record merge; `tests/test_config_listener_lifecycle.py` covers the lazy registration and the auto-close exclusion.
 - `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. `tests/test_stream_lifecycle.py` covers the add/remove/re-add race, duplicate `listen()` calls, cancellation while blocked reading content, close-during-connect, and close-during-backoff.
 
 ## Maintaining this file

teslemetry_stream/const.py

@@ -23,6 +23,7 @@ class Key(StrEnum):
     ERRORS = "errors"
     VEHICLE_DATA = "vehicle_data"
     STATE = "state"
+    CONFIG = "config"
     STATUS = "status"
     NETWORK_INTERFACE = "networkInterface"
     SITE_ID = "site_id"

teslemetry_stream/stream.py

@@ -66,7 +66,8 @@ def __init__(
         else:
             self.topics = None
         self._listeners: dict[
-            Callable[..., Any], tuple[Callable[[dict[str, Any]], None], dict[str, Any] | None]
+            Callable[..., Any],
+            tuple[Callable[[dict[str, Any]], None], dict[str, Any] | None, bool],
         ] = {}
         self._connection_listeners: dict[Callable[..., Any], Callable[[bool], None]] = {}
         self._listen_task: asyncio.Task[None] | None = None
@@ -361,13 +362,21 @@ async def __anext__(self) -> dict[str, Any]:
         raise StopAsyncIteration
 
     def async_add_listener(
-        self, callback: Callable[[dict[str, Any]], None], filters: dict[str, Any] | None = None
+        self,
+        callback: Callable[[dict[str, Any]], None],
+        filters: dict[str, Any] | None = None,
+        internal: bool = False,
     ) -> Callable[[], None]:
         """
         Listen for data updates.
 
         :param callback: Callback function to handle updates.
         :param filters: Filters to apply to the updates.
+        :param internal: True for a listener that keeps the client's own
+            state fresh (e.g. a vehicle's config-sync listener) rather than
+            serving a consumer callback. Excluded from the "last listener
+            removed" auto-close check, so a bookkeeping-only listener can't
+            pin the connection open forever once every real listener is gone.
         :return: Function to remove the listener.
         """
         schedule_refresh = not self._listeners
@@ -377,11 +386,11 @@ def remove_listener() -> None:
             Remove update listener.
             """
             self._listeners.pop(remove_listener)
-            if not self._listeners:
+            if not any(not is_internal for _, _, is_internal in self._listeners.values()):
                 LOGGER.info("Shutting down stream as there are no more listeners")
                 self.close()
 
-        self._listeners[remove_listener] = (callback, filters)
+        self._listeners[remove_listener] = (callback, filters, internal)
 
         # This is the first listener - start the owned listen task, unless
         # one is already running or manual mode delegates that to the caller.
@@ -415,7 +424,7 @@ async def listen(self) -> None:
         try:
             async for event in self:
                 if event:
-                    for listener, filters in self._listeners.values():
+                    for listener, filters, _internal in self._listeners.values():
                         if recursive_match(filters, event):
                             try:
                                 listener(event)

teslemetry_stream/vehicle.py

@@ -86,6 +86,11 @@ def __init__(self, stream: TeslemetryStream, vin: str):
         # Callers that arrive while it is running merge into `_config` and
         # await it instead of starting their own PATCH.
         self._flight = None
+        # Registration is deferred to first use (see _ensure_config_listener) -
+        # eagerly calling async_add_listener here would create the stream's
+        # owned task via asyncio.create_task before a loop necessarily exists,
+        # breaking a synchronous `TeslemetryStream(vin=...)`/vehicle construction.
+        self._config_listener_registered = False
 
     @property
     def config(self) -> dict[str, Any]:
@@ -115,13 +120,68 @@ async def get_config(self) -> None:
 
         req.raise_for_status()
 
+    def _ensure_config_listener(self) -> None:
+        """Register the internal config-sync listener once, lazily.
+
+        Keeps `fields`/`preferTyped` current when the server applies a
+        config change outside this client (another client, the console,
+        a Teslemetry-side migration), so add_field/prefer_typed's no-op
+        checks don't act on stale history. Marked `internal` so it doesn't
+        pin the stream open once every public listener is removed.
+        """
+        if self._config_listener_registered:
+            return
+        self._config_listener_registered = True
+        self.stream.async_add_listener(
+            self._on_config_event,
+            {Key.VIN: self.vin, Key.CONFIG: None},
+            internal=True,
+        )
+
+    def _on_config_event(self, event: dict[str, Any]) -> None:
+        """Sync the record from a server-pushed config event.
+
+        Only well-typed pieces are applied; a bad piece is logged and
+        skipped so it can't corrupt the last-known-good record, and the
+        other piece (if well-typed) still applies.
+        """
+        config = event.get(Key.CONFIG)
+        if not isinstance(config, dict):
+            LOGGER.warning(
+                "Ignoring malformed config event for %s: %r", self.vin, config
+            )
+            return
+
+        if "fields" in config:
+            fields = config["fields"]
+            if isinstance(fields, dict):
+                self.fields = fields
+            else:
+                LOGGER.warning(
+                    "Ignoring malformed fields in config event for %s: %r",
+                    self.vin,
+                    fields,
+                )
+
+        if "prefer_typed" in config:
+            prefer_typed = config["prefer_typed"]
+            if isinstance(prefer_typed, bool):
+                self.preferTyped = prefer_typed
+            else:
+                LOGGER.warning(
+                    "Ignoring malformed prefer_typed in config event for %s: %r",
+                    self.vin,
+                    prefer_typed,
+                )
+
     async def update_config(self, config: dict[str, Any]) -> None:
         """Request a configuration update for the vehicle.
 
         Merges into the pending desired config and joins the single in-flight
         flush for this vehicle rather than starting a new one, so that a
         batch of listeners scheduled at the same time produces one PATCH.
         """
+        self._ensure_config_listener()
 
         async with self.lock:
             self._config = merge(config, self._config)
@@ -231,6 +291,7 @@ async def post_config(self, config: dict[str, Any]) -> dict[str, Any]:
 
     async def add_field(self, field: Signal | str, interval: int | None = None) -> None:
         """Handle vehicle data from the stream."""
+        self._ensure_config_listener()
         if isinstance(field, Signal):
             field = field.value
 
@@ -249,6 +310,7 @@ async def add_field(self, field: Signal | str, interval: int | None = None) -> N
 
     async def prefer_typed(self, prefer_typed: bool) -> None:
         """Set prefer typed."""
+        self._ensure_config_listener()
         if self.preferTyped == prefer_typed:
             return
         await self.update_config({"prefer_typed": prefer_typed})

tests/test_batch_retry_storm.py

@@ -31,6 +31,11 @@ class FakeStream:
 
     manual = True
 
+    def async_add_listener(
+        self, callback: Any, filters: dict[str, Any] | None = None, internal: bool = False
+    ) -> Any:
+        return lambda: None
+
 
 def make_vehicle(vin: str, responses: list[Any]) -> TeslemetryStreamVehicle:
     """Build a vehicle that records payloads and replays canned responses."""