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: `__init__` unconditionally 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`. Because this listener is registered at construction, any stream stand-in passed to `TeslemetryStreamVehicle` (test doubles included) must implement `async_add_listener`. `tests/test_config_events.py` covers the record update, the resulting no-op/send behavior in `add_field`, and malformed/partial events.
 - `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/vehicle.py

@@ -86,6 +86,13 @@ 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
+        # 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.
+        self.stream.async_add_listener(
+            self._on_config_event, {Key.VIN: self.vin, Key.CONFIG: None}
+        )
 
     @property
     def config(self) -> dict[str, Any]:
@@ -115,6 +122,42 @@ 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"]
+            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.
 

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
+    ) -> Any:
+        return lambda: None
+
 
 def make_vehicle(vin: str, responses: list[Any]) -> TeslemetryStreamVehicle:
     """Build a vehicle that records payloads and replays canned responses."""

tests/test_config_events.py

@@ -0,0 +1,186 @@
+"""Config-update SSE events keep the internal config record fresh.
+
+The server pushes a ``config`` event (``Key.CONFIG`` / ``SseTopic.CONFIG``)
+shaped like ``{"vin": ..., "config": {"fields": {...}, "prefer_typed": bool}}``,
+mirroring the REST ``get_config`` response body.  ``TeslemetryStreamVehicle``
+registers an internal listener for it at construction so ``fields``/
+``preferTyped`` - and therefore the ``add_field``/``prefer_typed`` no-op
+checks - reflect current server truth rather than only what this client has
+itself requested or observed at connect.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any, Callable
+
+from teslemetry_stream.const import Key
+from teslemetry_stream.vehicle import TeslemetryStreamVehicle
+
+VIN = "TESTVIN0000000001"
+
+
+class FakeStream:
+    """Minimal stand-in for TeslemetryStream that captures the config listener."""
+
+    manual = True
+
+    def __init__(self) -> None:
+        self.config_listener: Callable[[dict[str, Any]], None] | None = None
+
+    def async_add_listener(
+        self,
+        callback: Callable[[dict[str, Any]], None],
+        filters: dict[str, Any] | None = None,
+    ) -> Callable[[], None]:
+        assert filters is not None
+        if Key.CONFIG in filters:
+            self.config_listener = callback
+        return lambda: None
+
+
+class CaptureWarnings(logging.Handler):
+    """Collect formatted WARNING records emitted by the library."""
+
+    def __init__(self) -> None:
+        super().__init__(level=logging.WARNING)
+        self.messages: list[str] = []
+
+    def emit(self, record: logging.LogRecord) -> None:
+        self.messages.append(record.getMessage())
+
+
+def check(label: str, ok: bool, detail: str = "") -> bool:
+    print(f"{label:<56} {'PASS' if ok else 'FAIL'}{'  ' + detail if detail else ''}")
+    return ok
+
+
+def make_vehicle() -> tuple[TeslemetryStreamVehicle, FakeStream]:
+    """Build a vehicle that records PATCH payloads instead of sending them."""
+    stream = FakeStream()
+    vehicle = TeslemetryStreamVehicle(stream, VIN)  # type: ignore[arg-type]
+    vehicle.sent = []  # type: ignore[attr-defined]
+
+    async def patch_config(config: dict[str, Any]) -> dict[str, Any]:
+        vehicle.sent.append(dict(config))  # type: ignore[attr-defined]
+        return {"updated_vehicles": 1}
+
+    vehicle.patch_config = patch_config  # type: ignore[assignment,method-assign]
+    return vehicle, stream
+
+
+async def main() -> None:
+    results = []
+
+    # A config event replaces the record with current server truth.
+    vehicle, stream = make_vehicle()
+    assert stream.config_listener is not None
+    stream.config_listener(
+        {
+            "vin": VIN,
+            "config": {
+                "fields": {"BatteryLevel": {"interval_seconds": 60}},
+                "prefer_typed": True,
+            },
+        }
+    )
+    results.append(
+        check(
+            "a config event updates fields and prefer_typed",
+            vehicle.fields == {"BatteryLevel": {"interval_seconds": 60}}
+            and vehicle.preferTyped is True,
+            f"fields {vehicle.fields}, prefer_typed {vehicle.preferTyped}",
+        )
+    )
+
+    # A request that now matches the updated record is skipped - no PATCH sent.
+    await vehicle.add_field("BatteryLevel", 60)
+    results.append(
+        check(
+            "a request matching the updated record is skipped",
+            vehicle.sent == [],
+            f"sent {vehicle.sent}",
+        )
+    )
+
+    # A request that differs from the updated record is still sent.
+    await vehicle.add_field("BatteryLevel", 30)
+    results.append(
+        check(
+            "a request differing from the updated record is sent",
+            len(vehicle.sent) == 1
+            and vehicle.sent[0]["fields"]["BatteryLevel"] == {"interval_seconds": 30},
+            f"sent {vehicle.sent}",
+        )
+    )
+
+    # A malformed/partial config event never corrupts the record.
+    handler = CaptureWarnings()
+    logger = logging.getLogger("teslemetry_stream")
+    logger.addHandler(handler)
+    try:
+        vehicle, stream = make_vehicle()
+        assert stream.config_listener is not None
+        stream.config_listener(
+            {
+                "vin": VIN,
+                "config": {
+                    "fields": {"BatteryLevel": {}},
+                    "prefer_typed": False,
+                },
+            }
+        )
+        good_fields, good_typed = dict(vehicle.fields), vehicle.preferTyped
+
+        # A non-dict "config" body is entirely rejected and logged.
+        stream.config_listener({"vin": VIN, "config": "not-a-dict"})
+        results.append(
+            check(
+                "a non-dict config event is ignored, logged, and keeps last-good",
+                vehicle.fields == good_fields
+                and vehicle.preferTyped == good_typed
+                and any("malformed" in m.lower() for m in handler.messages),
+                f"fields {vehicle.fields}, warnings {handler.messages}",
+            )
+        )
+
+        # A partially malformed body applies the well-typed piece and keeps
+        # the last-good value for the malformed piece.
+        handler.messages.clear()
+        stream.config_listener(
+            {"vin": VIN, "config": {"fields": "not-a-dict", "prefer_typed": True}}
+        )
+        results.append(
+            check(
+                "a partial config event applies the good field, keeps the bad one",
+                vehicle.fields == good_fields
+                and vehicle.preferTyped is True
+                and any("malformed" in m.lower() for m in handler.messages),
+                f"fields {vehicle.fields}, prefer_typed {vehicle.preferTyped}, "
+                f"warnings {handler.messages}",
+            )
+        )
+
+        # A config event missing a key entirely leaves that piece untouched.
+        handler.messages.clear()
+        stream.config_listener(
+            {"vin": VIN, "config": {"fields": {"CarType": {}}}}
+        )
+        results.append(
+            check(
+                "a config event omitting prefer_typed leaves it unchanged",
+                vehicle.fields == {"CarType": {}} and vehicle.preferTyped is True,
+                f"fields {vehicle.fields}, prefer_typed {vehicle.preferTyped}",
+            )
+        )
+    finally:
+        logger.removeHandler(handler)
+
+    print("-" * 72)
+    print("ALL PASS" if all(results) else "FAILURES PRESENT")
+    if not all(results):
+        raise SystemExit(1)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

tests/test_config_update.py

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

tests/test_field_type_coercion.py

@@ -31,10 +31,13 @@ def async_add_listener(
         callback: Callable[[dict[str, Any]], None],
         filters: dict[str, Any] | None = None,
     ) -> Callable[[], None]:
-        # filters carries {"vin": ..., "data": {Signal: None}} — grab the field
+        # filters carries {"vin": ..., "data": {Signal: None}} — grab the field.
+        # The vehicle's own internal config-sync listener has no "data" key;
+        # it's not under test here, so just ignore it.
         assert filters is not None
-        signal = next(iter(filters["data"]))
-        self.captured[signal] = callback
+        if "data" in filters:
+            signal = next(iter(filters["data"]))
+            self.captured[signal] = callback
         return lambda: None