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

AGENTS.md

@@ -16,7 +16,11 @@ 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.
-- `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.
+- `TeslemetryStreamVehicle` keeps `fields`/`preferTyped` fresh against server-side changes (another client, the console, a Teslemetry migration), not just this client's own history. `__init__` 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, unconditionally at construction - not lazily - so no connection can ever predate it and miss an event. A well-typed `fields`/`prefer_typed` piece replaces the corresponding record field (every nested `fields` entry must itself be a dict - one bad entry, e.g. a null, rejects the whole `fields` piece rather than leaving something `add_field` would later crash on); a missing piece is left untouched; a malformed piece is logged and skipped without touching the other piece or the pending `_config`. The stored `fields` dict (and each nested per-field dict) is copied, never the same object handed to public listeners for that same event - a consumer mutating its event in place must not corrupt the record.
+- `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)`.
+- The config-sync listener can only observe a server-side change while connected AND while the `config` topic isn't filtered out via `TeslemetryStream(topics=...)`. `add_field`/`prefer_typed`'s no-op skip is gated on `_record_is_live()` (both conditions true) rather than on `fields`/`preferTyped` alone - when not live, they send unconditionally instead of trying to force the record fresh, matching the pre-feature status quo (worst case one redundant PATCH, which the server handles fine). An earlier attempt forced a REST `get_config()` refresh before the check whenever disconnected; that was reverted - it added a failure path and could storm the API with one GET per caller in a batch. Test doubles need `connected` and `topics` attributes (`True`/`None` keep the record "live", matching pre-existing test expectations) for this reason.
+- `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 all three `_record_is_live()` states (disconnected, topic-filtered, live); `tests/test_stream_lifecycle.py` covers the internal-listener-survives-close restart case.
+- `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. `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, and internal-before-public dispatch order.
 
 ## 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,30 +362,50 @@ 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 both the "first
+            listener" start check and the "last listener removed" auto-close
+            check - on either side of the registry, only public listeners
+            count - so a bookkeeping-only listener can neither pin the
+            connection open forever nor, by itself, block a later public
+            listener from restarting a closed one.
         :return: Function to remove the listener.
         """
-        schedule_refresh = not self._listeners
+
+        def has_public_listener() -> bool:
+            return any(not is_internal for _, _, is_internal in self._listeners.values())
+
+        # A transition from zero to one *public* listeners, not merely a
+        # non-empty registry - an internal listener surviving a prior
+        # auto-close must not block a later public listener from restarting
+        # the owned task.
+        schedule_refresh = not internal and not has_public_listener()
 
         def remove_listener() -> None:
             """
             Remove update listener.
             """
             self._listeners.pop(remove_listener)
-            if not self._listeners:
+            if not has_public_listener():
                 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.
+        # This is the first public listener - start the owned listen task,
+        # unless one is already running or manual mode delegates that to the
+        # caller.
         if (
             schedule_refresh
             and not self.manual
@@ -415,7 +436,15 @@ async def listen(self) -> None:
         try:
             async for event in self:
                 if event:
-                    for listener, filters in self._listeners.values():
+                    # A snapshot, not a live view - a callback that creates a
+                    # vehicle (get_vehicle) or otherwise adds a listener
+                    # mid-dispatch must not mutate _listeners while this is
+                    # iterating it, which would raise RuntimeError and kill
+                    # the loop. Internal (bookkeeping) listeners go first, so
+                    # one can cache from the pristine event before any public
+                    # callback gets a chance to mutate it in place.
+                    ordered = sorted(self._listeners.values(), key=lambda item: not item[2])
+                    for listener, filters, _internal in ordered:
                         if recursive_match(filters, event):
                             try:
                                 listener(event)

teslemetry_stream/vehicle.py

@@ -46,6 +46,7 @@
     ShiftState,
     Signal,
     SpeedAssistLevel,
+    SseTopic,
     State,
     Status,
     SunroofInstalledState,
@@ -86,6 +87,16 @@ 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
+        # Registered from birth, not lazily, so no connection can ever
+        # predate this listener and miss a config event. Safe outside a
+        # running loop: `internal=True` makes async_add_listener's
+        # schedule_refresh unconditionally False, so it never reaches the
+        # asyncio.create_task() call that requires one.
+        self.stream.async_add_listener(
+            self._on_config_event,
+            {Key.VIN: self.vin, Key.CONFIG: None},
+            internal=True,
+        )
 
     @property
     def config(self) -> dict[str, Any]:
@@ -115,6 +126,51 @@ async def get_config(self) -> None:
 
         req.raise_for_status()
 
+    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"]
+            # Every entry must itself be a dict (e.g. {"interval_seconds": 60}
+            # or {}) - `fields: dict[str, dict[str, int]]` - so a downstream
+            # `self.fields[field].get(...)` (add_field's no-op check) can't
+            # raise AttributeError on a null/scalar entry that snuck in.
+            if isinstance(fields, dict) and all(
+                isinstance(value, dict) for value in fields.values()
+            ):
+                # Copied, not aliased - the event dict is also handed to
+                # public listeners, and a consumer mutating it in place
+                # must not corrupt this record.
+                self.fields = {field: dict(value) for field, value in fields.items()}
+            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.
 
@@ -234,8 +290,10 @@ async def add_field(self, field: Signal | str, interval: int | None = None) -> N
         if isinstance(field, Signal):
             field = field.value
 
-        if field in self.fields and (
-            interval is None or self.fields[field].get("interval_seconds") == interval
+        if (
+            self._record_is_live()
+            and field in self.fields
+            and (interval is None or self.fields[field].get("interval_seconds") == interval)
         ):
             LOGGER.debug(
                 "Streaming field %s already enabled @ %ss",
@@ -249,10 +307,26 @@ 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."""
-        if self.preferTyped == prefer_typed:
+        if self._record_is_live() and self.preferTyped == prefer_typed:
             return
         await self.update_config({"prefer_typed": prefer_typed})
 
+    def _record_is_live(self) -> bool:
+        """Whether the record is being kept current and can gate the no-op skip.
+
+        The skip is purely an optimization - the server handles a redundant
+        PATCH fine - so this only needs to answer "is the config-sync
+        listener actually able to observe a server-side change right now",
+        not force the record fresh. That requires both a live connection and
+        the `config` topic not being filtered out via `TeslemetryStream
+        (topics=...)`; if either is false, add_field/prefer_typed skip the
+        no-op check and always send, same as the pre-feature status quo.
+        """
+        if not self.stream.connected:
+            return False
+        topics = self.stream.topics
+        return topics is None or SseTopic.CONFIG in topics
+
     def _enable_field(self, field: Signal) -> None:
         """Enable a field for streaming from a listener."""
         asyncio.create_task(self.add_field(field))

tests/test_batch_retry_storm.py

@@ -30,6 +30,15 @@ class FakeStream:
     """Minimal stand-in for TeslemetryStream."""
 
     manual = True
+    # Keeps the record "live" (see _record_is_live) so add_field/prefer_typed's
+    # no-op check runs - these tests exercise the write path itself.
+    connected = True
+    topics = None
+
+    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: