fix: snapshot _connection_listeners before dispatch, not the live dict

AGENTS.md

@@ -20,7 +20,7 @@ This file is the project's committed home for project-intrinsic agent knowledge:
 - `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.
+- `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. `_update_connection_listeners()` has the same hazard for `_connection_listeners` (a connection listener calling `get_vehicle()` registers that vehicle's own connection listener mid-dispatch) and is fixed the same way, over a plain `list(...)` snapshot - no ordering requirement there, since connection listeners don't share a mutable event object. `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, internal-before-public dispatch order, and a connection listener mutating `_connection_listeners` mid-dispatch.
 
 ## Maintaining this file
 

teslemetry_stream/stream.py

@@ -224,8 +224,13 @@ def remove_listener() -> None:
 
  def _update_connection_listeners(self, value: bool | None = None) -> None:
  """Update all connection listeners with retry count"""
- for listener in self._connection_listeners.values():
- listener(self.connected if value is None else value)
+ # A snapshot, not a live view - a callback that creates a vehicle
+ # (get_vehicle) or otherwise adds a connection listener mid-dispatch
+ # must not mutate _connection_listeners while this is iterating it,
+ # which would raise RuntimeError and kill connect()/disconnect.
+ connected = self.connected if value is None else value
+ for listener in list(self._connection_listeners.values()):
+ listener(connected)
 
  async def connect(self) -> None:
  """

tests/test_stream_lifecycle.py

@@ -512,6 +512,51 @@ async def get_config() -> None:
  await asyncio.sleep(0)
 
 
+async def test_connect_survives_connection_listener_creating_vehicle_mid_dispatch(
+ results: list[bool],
+) -> None:
+ """A connection listener that calls get_vehicle() for an uncached VIN -
+ e.g. an integration reacting to reconnect by discovering a vehicle -
+ registers a new connection listener (TeslemetryStreamVehicle.__init__)
+ while _update_connection_listeners() is iterating _connection_listeners.
+ That must not raise and kill connect()."""
+ session = FakeSession()
+ stream = make_stream(session, manual=True)
+
+ discovered: list[TeslemetryStreamVehicle] = []
+
+ def on_connect(connected: bool) -> None:
+ if connected and not discovered:
+ discovered.append(stream.get_vehicle("NEWVIN1"))
+
+ stream.async_add_connection_listener(on_connect)
+
+ try:
+ await stream.connect()
+ connect_ok = True
+ except RuntimeError as error:
+ connect_ok = False
+ connect_error = repr(error)
+ results.append(
+ check(
+ "connect() survives a connection listener mutating _connection_listeners",
+ connect_ok,
+ "" if connect_ok else connect_error,
+ )
+ )
+ results.append(check("the listener discovered the new vehicle", len(discovered) == 1))
+ results.append(
+ check(
+ "the new vehicle's own connection listener is registered",
+ len(stream._connection_listeners) == 2,
+ f"connection listeners {len(stream._connection_listeners)}",
+ )
+ )
+
+ stream.close()
+ await asyncio.sleep(0)
+
+
 async def main() -> None:
  results: list[bool] = []
  await test_add_remove_readd_before_loop_runs(results)
@@ -523,6 +568,7 @@ async def main() -> None:
  await test_dispatch_survives_listener_creating_vehicle_mid_iteration(results)
  await test_internal_listener_sees_event_before_public_mutator(results)
  await test_vehicle_discovered_mid_dispatch_fetches_correctly_on_first_use(results)
+ await test_connect_survives_connection_listener_creating_vehicle_mid_dispatch(results)
 
  print("-" * 72)
  print("ALL PASS" if all(results) else "FAILURES PRESENT")