fix: harden SSE stream lifecycle (owned listen task, connect single-flight, real close)

AGENTS.md

@@ -16,6 +16,7 @@ This file is the project's committed home for project-intrinsic agent knowledge:
 - `site_info` events no longer carry `tariff_content`/`tariff_content_v2` (Teslemetry/api PR 318); 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), so it can't actually promise the whole REST-shaped document; 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.
 - `TeslemetryStream(topics=...)` (Teslemetry/api PR 319) 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 (must stay in sync with the api's `SSE_TOPICS`), 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 - and existing callers that never pass it are unaffected. 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.
+- `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/stream.py

@@ -69,6 +69,10 @@ def __init__(
             Callable[..., Any], tuple[Callable[[dict[str, Any]], None], dict[str, Any] | None]
         ] = {}
         self._connection_listeners: dict[Callable[..., Any], Callable[[bool], None]] = {}
+        self._listen_task: asyncio.Task[None] | None = None
+        # Created lazily in connect() - asyncio.Lock() requires a running
+        # loop on Python 3.9, and streams are commonly built before one.
+        self._connect_lock: asyncio.Lock | None = None
         self._session = session
         self.access_token = access_token
         self.parse_timestamp = parse_timestamp
@@ -230,44 +234,72 @@ async def connect(self) -> None:
         if not self.server:
             await self.get_config()
 
-        LOGGER.debug("Connecting to %s", self.server)
-        url = f"https://{self.server}/sse"
-        if self.vin:
-            url += f"/{self.vin}"
-        headers = await self.headers()
-        params = {"topics": ",".join(self.topics)} if self.topics else None
-        self._response = await self._session.get(
-            url,
-            headers=headers,
-            params=params,
-            raise_for_status=True,
-            timeout=aiohttp.ClientTimeout(
-                connect=5, sock_connect=5, sock_read=30, total=None
-            ),
-            chunked=True,
-        )
-        LOGGER.debug(
-            "Connected to %s with status %s", self._response.url, self._response.status
-        )
-        self.retries = 0
-        self._update_connection_listeners(True)
+        if self._connect_lock is None:
+            self._connect_lock = asyncio.Lock()
+        async with self._connect_lock:
+            if not self.active:
+                # Stopped while waiting for the lock; a concurrent caller may
+                # already be connected, or a stop was requested outright.
+                return
+
+            LOGGER.debug("Connecting to %s", self.server)
+            url = f"https://{self.server}/sse"
+            if self.vin:
+                url += f"/{self.vin}"
+            headers = await self.headers()
+            params = {"topics": ",".join(self.topics)} if self.topics else None
+            response = await self._session.get(
+                url,
+                headers=headers,
+                params=params,
+                raise_for_status=True,
+                timeout=aiohttp.ClientTimeout(
+                    connect=5, sock_connect=5, sock_read=30, total=None
+                ),
+                chunked=True,
+            )
+            if not self.active:
+                # Stopped while the request was in flight - discard it rather
+                # than publish a response nobody will ever close.
+                response.close()
+                return
+            if self._response is not None:
+                self._response.close()
+            self._response = response
+            LOGGER.debug(
+                "Connected to %s with status %s", self._response.url, self._response.status
+            )
+            self.retries = 0
+            self._update_connection_listeners(True)
 
     def disconnect(self) -> None:
         """
         Disconnect from the telemetry stream.
         """
-        self.active = False
         self.close()
 
-    def close(self) -> None:
+    def _close_response(self) -> None:
         """
-        Close connection.
+        Close the current response, if any, without changing whether the
+        stream is meant to keep running - used by reconnect paths and by
+        listen()'s cleanup, as opposed to close()'s full stop.
         """
         if self._response is not None:
             LOGGER.debug("Disconnecting from %s", self.server)
             self._response.close()
             self._response = None
-        self._update_connection_listeners(False)
+            self._update_connection_listeners(False)
+
+    def close(self) -> None:
+        """
+        Stop the stream: closes the response and cancels the owned listen
+        task so a running listener does not immediately reconnect.
+        """
+        self.active = False
+        task, self._listen_task = self._listen_task, None
+        if task is not None and not task.done():
+            task.cancel()
+        self._close_response()
 
     def __aiter__(self) -> TeslemetryStream:
         """
@@ -312,17 +344,17 @@ async def __anext__(self) -> dict[str, Any]:
                 raise e
             except TeslemetryStreamEnded:
                 LOGGER.warning("Stream ended by server")
-                self.close()
+                self._close_response()
             except aiohttp.ClientError as error:
                 LOGGER.warning("Client error: %s", repr(error))
-                self.close()
+                self._close_response()
                 delay = min(2**self.retries, 600)
                 LOGGER.debug("Reconnecting in %s seconds", delay)
                 await asyncio.sleep(delay)
                 self.retries += 1
             except Exception as error:
                 LOGGER.error("Unexpected error: %s", repr(error))
-                self.close()
+                self._close_response()
                 LOGGER.debug("Reconnecting in %s seconds", 1)
                 await asyncio.sleep(1)
 
@@ -347,28 +379,52 @@ def remove_listener() -> None:
             self._listeners.pop(remove_listener)
             if not self._listeners:
                 LOGGER.info("Shutting down stream as there are no more listeners")
-                self.active = False
+                self.close()
 
         self._listeners[remove_listener] = (callback, filters)
 
-        # This is the first listener, set up task.
-        if schedule_refresh and not self.manual:
-            asyncio.create_task(self.listen())
+        # This is the first listener - start the owned listen task, unless
+        # one is already running or manual mode delegates that to the caller.
+        if (
+            schedule_refresh
+            and not self.manual
+            and (self._listen_task is None or self._listen_task.done())
+        ):
+            self._listen_task = asyncio.create_task(self.listen())
 
         return remove_listener
 
     async def listen(self) -> None:
         """
         Listen to the telemetry stream.
-        """
-        async for event in self:
-            if event:
-                for listener, filters in self._listeners.values():
-                    if recursive_match(filters, event):
-                        try:
-                            listener(event)
-                        except Exception as error:
-                            LOGGER.error("Uncaught error in listener: %s", error)
+
+        A second concurrent call joins the already-running owned task
+        instead of starting a competing reader on the same connection.
+        """
+        current_task = asyncio.current_task()
+        existing_task = self._listen_task
+        if (
+            existing_task is not None
+            and not existing_task.done()
+            and existing_task is not current_task
+        ):
+            await existing_task
+            return
+
+        self._listen_task = current_task
+        try:
+            async for event in self:
+                if event:
+                    for listener, filters in self._listeners.values():
+                        if 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:
+                self._listen_task = None
         LOGGER.debug("Listen has finished")
 
     def listen_Credits(