fix: gate config no-op skip on a populated flag, not connection state

AGENTS.md

@@ -17,9 +17,9 @@ This file is the project's committed home for project-intrinsic agent knowledge:
 - 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__` 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.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`/`prefer_typed` gate their 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`/`preferTyped` outright with no network call. `_populated` is set by a successful `get_config()` (200 or 404 - both are an authoritative answer) 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.
+- `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` 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/vehicle.py

@@ -46,7 +46,6 @@
  ShiftState,
  Signal,
  SpeedAssistLevel,
- SseTopic,
  State,
  Status,
  SunroofInstalledState,
@@ -73,6 +72,8 @@ class TeslemetryStreamVehicle:
  preferTyped: bool | None
  _config: dict[str, Any]
  _flight: asyncio.Task[None] | None
+ _populated: bool
+ _populate_flight: asyncio.Task[None] | None
 
  def __init__(self, stream: TeslemetryStream, vin: str):
  # A dictionary of TelemetryField keys and null values
@@ -83,10 +84,18 @@ def __init__(self, stream: TeslemetryStream, vin: str):
  self.fields = {}
  self.preferTyped = None
  self._config = {}
+ # Whether fields/preferTyped reflect a real server answer (a push
+ # event or a REST fetch) rather than just their unset defaults -
+ # gates add_field/prefer_typed's lazy REST fetch below.
+ self._populated = False
  # The single in-flight (or most recently completed) coalesced flush.
  # Callers that arrive while it is running merge into `_config` and
  # await it instead of starting their own PATCH.
  self._flight = None
+ # The single in-flight populating get_config() call, so a batch of
+ # listeners (e.g. HA integration setup) discovering an unpopulated
+ # vehicle joins one GET instead of each starting its own.
+ self._populate_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
@@ -97,6 +106,11 @@ def __init__(self, stream: TeslemetryStream, vin: str):
  {Key.VIN: self.vin, Key.CONFIG: None},
  internal=True,
  )
+ # A disconnect can leave fields/preferTyped stale until the next
+ # connection's config snapshot arrives - unpopulate so a
+ # field-config call landing in that window awaits a fresh fetch
+ # instead of trusting the pre-disconnect record.
+ self.stream.async_add_connection_listener(self._on_connection_event)
 
  @property
  def config(self) -> dict[str, Any]:
@@ -120,12 +134,37 @@ async def get_config(self) -> None:
 
  self.fields = response.get("fields", {})
  self.preferTyped = response.get("prefer_typed", False)
+ self._populated = True
  return
  if req.status == 404:
+ # No config exists for this vehicle yet - an authoritative
+ # answer (empty), not a missing one.
+ self._populated = True
  return
 
  req.raise_for_status()
 
+ def _on_connection_event(self, connected: bool) -> None:
+ """Unpopulate on disconnect - see the __init__ registration comment."""
+ if not connected:
+ self._populated = False
+
+ async def _ensure_populated(self) -> None:
+ """Lazily fetch current config over REST if not yet known.
+
+ Optimistic updates from `_on_config_event` keep the record fresh
+ once established; this only covers the gap before a connection's
+ first snapshot (or after a disconnect) arrives. Concurrent callers
+ join the same fetch rather than each issuing their own GET.
+ """
+ if self._populated:
+ return
+ flight = self._populate_flight
+ if flight is None or flight.done():
+ flight = asyncio.ensure_future(self.get_config())
+ self._populate_flight = flight
+ await asyncio.shield(flight)
+
  def _on_config_event(self, event: dict[str, Any]) -> None:
  """Sync the record from a server-pushed config event.
 
@@ -140,6 +179,8 @@ def _on_config_event(self, event: dict[str, Any]) -> None:
  )
  return
 
+ self._populated = True
+
  if "fields" in config:
  fields = config["fields"]
  # Every entry must itself be a dict (e.g. {"interval_seconds": 60}
@@ -290,10 +331,10 @@ async def add_field(self, field: Signal | str, interval: int | None = None) -> N
  if isinstance(field, Signal):
  field = field.value
 
-if (
- self._record_is_live()
- and field in self.fields
-and (interval is None or self.fields[field].get("interval_seconds") == interval)
+await self._ensure_populated()
+
+if field in self.fields and (
+ interval is None or self.fields[field].get("interval_seconds") == interval
  ):
  LOGGER.debug(
  "Streaming field %s already enabled @ %ss",
@@ -307,26 +348,11 @@ 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._record_is_live() and self.preferTyped == prefer_typed:
+ await self._ensure_populated()
+ if 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,20 +30,23 @@ 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 async_add_connection_listener(self, callback: Any) -> Any:
+ return lambda: None
+
 
 def make_vehicle(vin: str, responses: list[Any]) -> TeslemetryStreamVehicle:
  """Build a vehicle that records payloads and replays canned responses."""
  vehicle = TeslemetryStreamVehicle(FakeStream(), vin) # type: ignore[arg-type]
+ # These tests exercise the write path, not the lazy-populate fetch (the
+ # fake stream has no REST session to serve one) - mark it populated like
+ # a real connection's config snapshot already would have.
+ vehicle._populated = True
  vehicle.sent = [] # type: ignore[attr-defined]
 
  async def patch_config(config: dict[str, Any]) -> dict[str, Any]:

tests/test_config_events.py

@@ -26,10 +26,6 @@ class FakeStream:
  """Minimal stand-in for TeslemetryStream that captures the config listener."""
 
  manual = True
- # Keeps the record "live" (see _record_is_live) so add_field/prefer_typed's
- # no-op check runs - these tests exercise the event-driven merge itself.
- connected = True
- topics = None
 
  def __init__(self) -> None:
  self.config_listener: Callable[[dict[str, Any]], None] | None = None
@@ -45,6 +41,11 @@ def async_add_listener(
  self.config_listener = callback
  return lambda: None
 
+ def async_add_connection_listener(
+ self, callback: Callable[[bool], None]
+ ) -> Callable[[], None]:
+ return lambda: None
+
 
 class CaptureWarnings(logging.Handler):
  """Collect formatted WARNING records emitted by the library."""