fix: close reconnect-window gap in add_field/prefer_typed no-op skip

AGENTS.md

@@ -18,8 +18,8 @@ This file is the project's committed home for project-intrinsic agent knowledge:
 - `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.
+- The config-sync listener can only observe a server-side change while connected, while the `config` topic isn't filtered out via `TeslemetryStream(topics=...)`, AND after this connection's own config snapshot has actually been applied. `add_field`/`prefer_typed`'s no-op skip is gated on `_record_is_live()` (all three 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. The third condition exists because `TeslemetryStream.connected` flips true as soon as `connect()` installs the response, before `listen()` has dispatched a single event - a connection listener reacting to reconnect can otherwise run `add_field`/`prefer_typed` in that window and match against the stale pre-disconnect record. `TeslemetryStream._connection_id` (bumped once per successful `connect()`) and `TeslemetryStreamVehicle._synced_connection_id` (set to it in `_on_config_event`) close that gap: `_record_is_live()` also requires the two to match. Both default to `0`/read via `getattr(..., 0)` so stream stand-ins that skip `_connection_id` entirely (most test doubles, which never call the real `connect()`) keep the pre-existing "live" behavior.
+- `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; `tests/test_reconnect_config_window.py` covers the reconnect-window race (connected-but-presnapshot no-op skip).
 - `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/stream.py

@@ -20,6 +20,10 @@ class TeslemetryStream:
  """Teslemetry Stream Client"""
 
  _response: aiohttp.ClientResponse | None = None
+ # Bumped each time connect() installs a new response - lets a vehicle
+ # tell "this connection's config snapshot has been applied" apart from
+ # "some past connection's was".
+ _connection_id: int = 0
 
  def __init__(
  self,
@@ -267,6 +271,7 @@ async def connect(self) -> None:
  if self._response is not None:
  self._response.close()
  self._response = response
+ self._connection_id += 1
  LOGGER.debug(
  "Connected to %s with status %s", self._response.url, self._response.status
  )

teslemetry_stream/vehicle.py

@@ -83,6 +83,11 @@ def __init__(self, stream: TeslemetryStream, vin: str):
  self.fields = {}
  self.preferTyped = None
  self._config = {}
+ # The connection id (see TeslemetryStream._connection_id) whose
+ # snapshot this record last reflects - gates _record_is_live so a
+ # reconnect can't be mistaken for "already synced" before its own
+ # config event has actually arrived.
+ self._synced_connection_id = getattr(stream, "_connection_id", 0)
  # 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.
@@ -140,6 +145,8 @@ def _on_config_event(self, event: dict[str, Any]) -> None:
  )
  return
 
+ self._synced_connection_id = getattr(self.stream, "_connection_id", 0)
+
  if "fields" in config:
  fields = config["fields"]
  # Every entry must itself be a dict (e.g. {"interval_seconds": 60}
@@ -317,13 +324,21 @@ def _record_is_live(self) -> bool:
  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.
+ not force the record fresh. That requires a live connection, the
+ `config` topic not being filtered out via `TeslemetryStream
+ (topics=...)`, AND this connection's own config snapshot having
+ already been applied - `connected` flips true as soon as the
+ response is installed, before that snapshot has been dispatched, so
+ without this last check a listener reacting to reconnect could match
+ against the pre-disconnect record and skip a PATCH that server-side
+ drift actually required. If any of the three 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
+ if getattr(self.stream, "_connection_id", 0) != self._synced_connection_id:
+ return False
  topics = self.stream.topics
  return topics is None or SseTopic.CONFIG in topics
 

tests/test_reconnect_config_window.py

@@ -0,0 +1,138 @@
+"""Regression test for a reconnect-window gap in the config no-op skip.
+
+``TeslemetryStream.connect()`` sets ``self._response`` and notifies
+connection listeners (``stream.connected`` becomes True) before ``listen()``
+has read a single SSE line - in particular, before the server's post-connect
+config snapshot has been dispatched to the vehicle's internal config
+listener. ``add_field``/``prefer_typed``'s no-op skip is gated on
+``_record_is_live()``, which only checks "connected and not topic-filtered" -
+not "has this connection's snapshot actually been applied yet". A connection
+listener that calls ``add_field()`` in that window can match against the
+stale pre-disconnect record and skip its PATCH, and nothing retries it once
+the snapshot later reveals the mismatch.
+"""
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+from teslemetry_stream.const import Key
+from teslemetry_stream.stream import TeslemetryStream
+from teslemetry_stream.vehicle import TeslemetryStreamVehicle
+
+VIN = "TESTVIN0000000001"
+
+
+class FakeResponse:
+ """A response whose content is never actually iterated in this test."""
+
+ status = 200
+ url = "https://api.teslemetry.com/sse"
+
+ def close(self) -> None:
+ pass
+
+
+class FakeSession:
+ """A session whose get() hands back a connected-but-empty response."""
+
+ async def get(self, url: str, **kwargs: Any) -> FakeResponse:
+ return FakeResponse()
+
+
+def check(label: str, ok: bool, detail: str = "") -> bool:
+ print(f"{label:<72} {'PASS' if ok else 'FAIL'}{' ' + detail if detail else ''}")
+ return ok
+
+
+def make_stream(**kwargs: Any) -> TeslemetryStream:
+ kwargs.setdefault("manual", True)
+ return TeslemetryStream(
+ session=FakeSession(), # type: ignore[arg-type]
+ access_token="test-token",
+ server="api.teslemetry.com",
+ **kwargs,
+ )
+
+
+def make_vehicle_with_capture(
+ stream: TeslemetryStream,
+) -> tuple[TeslemetryStreamVehicle, list[dict[str, Any]]]:
+ vehicle = TeslemetryStreamVehicle(stream, VIN)
+ sent: list[dict[str, Any]] = []
+
+ async def patch_config(config: dict[str, Any]) -> dict[str, Any]:
+ sent.append(dict(config))
+ return {"updated_vehicles": 1}
+
+ vehicle.patch_config = patch_config # type: ignore[assignment,method-assign]
+ return vehicle, sent
+
+
+async def test_reconnect_window_add_field_matches_stale_record(
+ results: list[bool],
+) -> None:
+ stream = make_stream()
+ vehicle, sent = make_vehicle_with_capture(stream)
+
+ # Pre-disconnect state: server had BatteryLevel @ 60s.
+ vehicle.fields = {"BatteryLevel": {"interval_seconds": 60}}
+ results.append(check("stream starts disconnected", not stream.connected))
+
+ # A connection listener reacting to reconnect - e.g. an HA integration
+ # re-asserting its desired fields - the exact scenario `_record_is_live`
+ # exists to protect against a stale match.
+ scheduled: list[asyncio.Task[None]] = []
+
+ def on_connect(connected: bool) -> None:
+ if connected:
+ scheduled.append(asyncio.ensure_future(vehicle.add_field("BatteryLevel", 60)))
+
+ stream.async_add_connection_listener(on_connect)
+
+ # Reconnect: this flips `connected` True and fires `on_connect`
+ # synchronously, scheduling (not yet running) the add_field task.
+ await stream.connect()
+ results.append(check("stream reports connected immediately after connect()", stream.connected))
+
+ # Let the scheduled add_field task run its no-op check BEFORE the
+ # server's config snapshot for this connection has arrived - the window
+ # under test. Server-side truth (unbeknownst to this client yet) is
+ # actually 30s, not the stale 60s the record still holds.
+ await asyncio.sleep(0)
+
+ # Now the snapshot for the new connection arrives.
+ vehicle._on_config_event(
+ {Key.VIN: VIN, Key.CONFIG: {"fields": {"BatteryLevel": {"interval_seconds": 30}}}}
+ )
+
+ await asyncio.gather(*scheduled)
+
+ results.append(
+ check(
+ "add_field sends a PATCH instead of matching the pre-reconnect stale record",
+ len(sent) == 1 and sent[0]["fields"]["BatteryLevel"] == {"interval_seconds": 60},
+ f"sent {sent}",
+ )
+ )
+ results.append(
+ check(
+ "the desired 60s interval is the vehicle's final state, not the stale 30s snapshot",
+ vehicle.fields.get("BatteryLevel") == {"interval_seconds": 60},
+ f"fields {vehicle.fields}",
+ )
+ )
+
+
+async def main() -> None:
+ results: list[bool] = []
+ await test_reconnect_window_add_field_matches_stale_record(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())