fix: seed a vehicle discovered mid-dispatch from its triggering config event

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)`.
 - 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.
+- `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. A listener added mid-dispatch (not in the original snapshot) still gets one extra pass at the *same* event once the snapshot's dispatch finishes - otherwise a vehicle created by `get_vehicle()` from within that event's dispatch would miss the very event that revealed its config and stay seeded empty. `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 mid-dispatch-created vehicle being seeded from its triggering event.
 
 ## Maintaining this file
 

teslemetry_stream/stream.py

@@ -443,13 +443,26 @@ async def listen(self) -> None:
                     # 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.
+                    dispatched_keys = set(self._listeners.keys())
                     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)
                             except Exception as error:
                                 LOGGER.error("Uncaught error in listener: %s", error)
+                    # A callback above may have added a listener (e.g.
+                    # get_vehicle() for an uncached VIN registers that
+                    # vehicle's internal config listener) that missed the
+                    # snapshot - give listeners added mid-dispatch one more
+                    # pass at this same event so a newly-created vehicle
+                    # isn't seeded stale.
+                    for key, (listener, filters, _internal) in list(self._listeners.items()):
+                        if key not in dispatched_keys and recursive_match(filters, event):
+                            try:
+                                listener(event)
+                            except Exception as error:
+                                LOGGER.error("Uncaught error in listener: %s", error)
         finally:
             self._close_response()
             if self._listen_task is current_task:

tests/test_stream_lifecycle.py

@@ -434,6 +434,59 @@ def public_mutator(event: dict[str, Any]) -> None:
     await asyncio.sleep(0)
 
 
+async def test_vehicle_discovered_mid_dispatch_seeds_from_current_config_event(
+    results: list[bool],
+) -> None:
+    """A generic public listener that discovers an uncached VIN and calls
+    get_vehicle() while handling that VIN's own config event registers a new
+    internal listener that the current dispatch already snapshotted past.
+    Without a second pass over listeners added mid-dispatch, the newly
+    created vehicle would miss the very event that revealed its config and
+    sit at fields={}/preferTyped=None until (if ever) another config event
+    arrives."""
+    new_vin = "TESTVIN0000000002"
+    session = FakeSession()
+    session.content_factory = lambda: FakeEventContent(
+        [
+            (
+                b'data: {"vin": "'
+                + new_vin.encode()
+                + b'", "config": {"fields": '
+                + b'{"BatteryLevel": {"interval_seconds": 60}}, '
+                + b'"prefer_typed": true}}\n'
+            )
+        ]
+    )
+    stream = make_stream(session)
+
+    discovered: list[TeslemetryStreamVehicle] = []
+
+    def generic_listener(event: dict[str, Any]) -> None:
+        vin = event.get("vin")
+        if vin and vin not in stream.vehicles:
+            discovered.append(stream.get_vehicle(vin))
+
+    stream.async_add_listener(generic_listener, {"vin": None})
+
+    for _ in range(5):
+        await asyncio.sleep(0)
+
+    results.append(check("the listener discovered the new vehicle", len(discovered) == 1))
+    if discovered:
+        vehicle = discovered[0]
+        results.append(
+            check(
+                "the newly discovered vehicle is seeded from the event that created it",
+                vehicle.fields.get("BatteryLevel") == {"interval_seconds": 60}
+                and vehicle.preferTyped is True,
+                f"fields {vehicle.fields}, prefer_typed {vehicle.preferTyped}",
+            )
+        )
+
+    stream.close()
+    await asyncio.sleep(0)
+
+
 async def main() -> None:
     results: list[bool] = []
     await test_add_remove_readd_before_loop_runs(results)
@@ -444,6 +497,7 @@ async def main() -> None:
     await test_restart_after_public_readd_with_internal_listener_present(results)
     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_seeds_from_current_config_event(results)
 
     print("-" * 72)
     print("ALL PASS" if all(results) else "FAILURES PRESENT")