diff --git a/AGENTS.md b/AGENTS.md
index 3e8600b..850cc50 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -16,7 +16,11 @@ This file is the project's committed home for project-intrinsic agent knowledge:
 - `site_info` events do not carry `tariff_content`/`tariff_content_v2`; 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); 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. `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.
-- `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.
+- `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` 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
 
diff --git a/teslemetry_stream/const.py b/teslemetry_stream/const.py
index 967744a..1fa300a 100644
--- a/teslemetry_stream/const.py
+++ b/teslemetry_stream/const.py
@@ -23,6 +23,7 @@ class Key(StrEnum):
     ERRORS = "errors"
     VEHICLE_DATA = "vehicle_data"
     STATE = "state"
+    CONFIG = "config"
     STATUS = "status"
     NETWORK_INTERFACE = "networkInterface"
     SITE_ID = "site_id"
diff --git a/teslemetry_stream/stream.py b/teslemetry_stream/stream.py
index c96fbc6..2c560b4 100644
--- a/teslemetry_stream/stream.py
+++ b/teslemetry_stream/stream.py
@@ -66,7 +66,8 @@ def __init__(
         else:
             self.topics = None
         self._listeners: dict[
-            Callable[..., Any], tuple[Callable[[dict[str, Any]], None], dict[str, Any] | None]
+            Callable[..., Any],
+            tuple[Callable[[dict[str, Any]], None], dict[str, Any] | None, bool],
         ] = {}
         self._connection_listeners: dict[Callable[..., Any], Callable[[bool], None]] = {}
         self._listen_task: asyncio.Task[None] | None = None
@@ -361,30 +362,50 @@ async def __anext__(self) -> dict[str, Any]:
         raise StopAsyncIteration
 
     def async_add_listener(
-        self, callback: Callable[[dict[str, Any]], None], filters: dict[str, Any] | None = None
+        self,
+        callback: Callable[[dict[str, Any]], None],
+        filters: dict[str, Any] | None = None,
+        internal: bool = False,
     ) -> Callable[[], None]:
         """
         Listen for data updates.
 
         :param callback: Callback function to handle updates.
         :param filters: Filters to apply to the updates.
+        :param internal: True for a listener that keeps the client's own
+            state fresh (e.g. a vehicle's config-sync listener) rather than
+            serving a consumer callback. Excluded from both the "first
+            listener" start check and the "last listener removed" auto-close
+            check - on either side of the registry, only public listeners
+            count - so a bookkeeping-only listener can neither pin the
+            connection open forever nor, by itself, block a later public
+            listener from restarting a closed one.
         :return: Function to remove the listener.
         """
-        schedule_refresh = not self._listeners
+
+        def has_public_listener() -> bool:
+            return any(not is_internal for _, _, is_internal in self._listeners.values())
+
+        # A transition from zero to one *public* listeners, not merely a
+        # non-empty registry - an internal listener surviving a prior
+        # auto-close must not block a later public listener from restarting
+        # the owned task.
+        schedule_refresh = not internal and not has_public_listener()
 
         def remove_listener() -> None:
             """
             Remove update listener.
             """
             self._listeners.pop(remove_listener)
-            if not self._listeners:
+            if not has_public_listener():
                 LOGGER.info("Shutting down stream as there are no more listeners")
                 self.close()
 
-        self._listeners[remove_listener] = (callback, filters)
+        self._listeners[remove_listener] = (callback, filters, internal)
 
-        # This is the first listener - start the owned listen task, unless
-        # one is already running or manual mode delegates that to the caller.
+        # This is the first public 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
@@ -415,7 +436,15 @@ async def listen(self) -> None:
         try:
             async for event in self:
                 if event:
-                    for listener, filters in self._listeners.values():
+                    # A snapshot, not a live view - a callback that creates a
+                    # vehicle (get_vehicle) or otherwise adds a listener
+                    # mid-dispatch must not mutate _listeners while this is
+                    # iterating it, which would raise RuntimeError and kill
+                    # 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.
+                    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)
diff --git a/teslemetry_stream/vehicle.py b/teslemetry_stream/vehicle.py
index 6bf9e7e..e28c5f6 100644
--- a/teslemetry_stream/vehicle.py
+++ b/teslemetry_stream/vehicle.py
@@ -46,6 +46,7 @@
     ShiftState,
     Signal,
     SpeedAssistLevel,
+    SseTopic,
     State,
     Status,
     SunroofInstalledState,
@@ -86,6 +87,16 @@ def __init__(self, stream: TeslemetryStream, vin: str):
         # Callers that arrive while it is running merge into `_config` and
         # await it instead of starting their own PATCH.
         self._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
+        # schedule_refresh unconditionally False, so it never reaches the
+        # asyncio.create_task() call that requires one.
+        self.stream.async_add_listener(
+            self._on_config_event,
+            {Key.VIN: self.vin, Key.CONFIG: None},
+            internal=True,
+        )
 
     @property
     def config(self) -> dict[str, Any]:
@@ -115,6 +126,51 @@ async def get_config(self) -> None:
 
         req.raise_for_status()
 
+    def _on_config_event(self, event: dict[str, Any]) -> None:
+        """Sync the record from a server-pushed config event.
+
+        Only well-typed pieces are applied; a bad piece is logged and
+        skipped so it can't corrupt the last-known-good record, and the
+        other piece (if well-typed) still applies.
+        """
+        config = event.get(Key.CONFIG)
+        if not isinstance(config, dict):
+            LOGGER.warning(
+                "Ignoring malformed config event for %s: %r", self.vin, config
+            )
+            return
+
+        if "fields" in config:
+            fields = config["fields"]
+            # Every entry must itself be a dict (e.g. {"interval_seconds": 60}
+            # or {}) - `fields: dict[str, dict[str, int]]` - so a downstream
+            # `self.fields[field].get(...)` (add_field's no-op check) can't
+            # raise AttributeError on a null/scalar entry that snuck in.
+            if isinstance(fields, dict) and all(
+                isinstance(value, dict) for value in fields.values()
+            ):
+                # Copied, not aliased - the event dict is also handed to
+                # public listeners, and a consumer mutating it in place
+                # must not corrupt this record.
+                self.fields = {field: dict(value) for field, value in fields.items()}
+            else:
+                LOGGER.warning(
+                    "Ignoring malformed fields in config event for %s: %r",
+                    self.vin,
+                    fields,
+                )
+
+        if "prefer_typed" in config:
+            prefer_typed = config["prefer_typed"]
+            if isinstance(prefer_typed, bool):
+                self.preferTyped = prefer_typed
+            else:
+                LOGGER.warning(
+                    "Ignoring malformed prefer_typed in config event for %s: %r",
+                    self.vin,
+                    prefer_typed,
+                )
+
     async def update_config(self, config: dict[str, Any]) -> None:
         """Request a configuration update for the vehicle.
 
@@ -234,8 +290,10 @@ async def add_field(self, field: Signal | str, interval: int | None = None) -> N
         if isinstance(field, Signal):
             field = field.value
 
-        if field in self.fields and (
-            interval is None or self.fields[field].get("interval_seconds") == interval
+        if (
+            self._record_is_live()
+            and field in self.fields
+            and (interval is None or self.fields[field].get("interval_seconds") == interval)
         ):
             LOGGER.debug(
                 "Streaming field %s already enabled @ %ss",
@@ -249,10 +307,26 @@ 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.preferTyped == prefer_typed:
+        if self._record_is_live() and 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))
diff --git a/tests/test_batch_retry_storm.py b/tests/test_batch_retry_storm.py
index 116a4d4..5655646 100644
--- a/tests/test_batch_retry_storm.py
+++ b/tests/test_batch_retry_storm.py
@@ -30,6 +30,15 @@ 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 make_vehicle(vin: str, responses: list[Any]) -> TeslemetryStreamVehicle:
diff --git a/tests/test_config_events.py b/tests/test_config_events.py
new file mode 100644
index 0000000..1737550
--- /dev/null
+++ b/tests/test_config_events.py
@@ -0,0 +1,254 @@
+"""Config-update SSE events keep the internal config record fresh.
+
+The server pushes a ``config`` event (``Key.CONFIG`` / ``SseTopic.CONFIG``)
+shaped like ``{"vin": ..., "config": {"fields": {...}, "prefer_typed": bool}}``,
+mirroring the REST ``get_config`` response body.  ``TeslemetryStreamVehicle``
+registers an internal listener for it at construction (see
+``test_config_listener_lifecycle.py`` for why that's safe even outside a
+running loop) so ``fields``/``preferTyped`` - and therefore the
+``add_field``/``prefer_typed`` no-op checks - reflect current server truth
+rather than only what this client has itself requested or observed at
+connect.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any, Callable
+
+from teslemetry_stream.const import Key
+from teslemetry_stream.vehicle import TeslemetryStreamVehicle
+
+VIN = "TESTVIN0000000001"
+
+
+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
+
+    def async_add_listener(
+        self,
+        callback: Callable[[dict[str, Any]], None],
+        filters: dict[str, Any] | None = None,
+        internal: bool = False,
+    ) -> Callable[[], None]:
+        assert filters is not None
+        if Key.CONFIG in filters:
+            self.config_listener = callback
+        return lambda: None
+
+
+class CaptureWarnings(logging.Handler):
+    """Collect formatted WARNING records emitted by the library."""
+
+    def __init__(self) -> None:
+        super().__init__(level=logging.WARNING)
+        self.messages: list[str] = []
+
+    def emit(self, record: logging.LogRecord) -> None:
+        self.messages.append(record.getMessage())
+
+
+def check(label: str, ok: bool, detail: str = "") -> bool:
+    print(f"{label:<56} {'PASS' if ok else 'FAIL'}{'  ' + detail if detail else ''}")
+    return ok
+
+
+def make_vehicle() -> tuple[TeslemetryStreamVehicle, FakeStream]:
+    """Build a vehicle that records PATCH payloads instead of sending them."""
+    stream = FakeStream()
+    vehicle = TeslemetryStreamVehicle(stream, VIN)  # type: ignore[arg-type]
+    vehicle.sent = []  # type: ignore[attr-defined]
+
+    async def patch_config(config: dict[str, Any]) -> dict[str, Any]:
+        vehicle.sent.append(dict(config))  # type: ignore[attr-defined]
+        return {"updated_vehicles": 1}
+
+    vehicle.patch_config = patch_config  # type: ignore[assignment,method-assign]
+    return vehicle, stream
+
+
+async def main() -> None:
+    results = []
+
+    # A config event replaces the record with current server truth.
+    vehicle, stream = make_vehicle()
+    assert stream.config_listener is not None
+    stream.config_listener(
+        {
+            "vin": VIN,
+            "config": {
+                "fields": {"BatteryLevel": {"interval_seconds": 60}},
+                "prefer_typed": True,
+            },
+        }
+    )
+    results.append(
+        check(
+            "a config event updates fields and prefer_typed",
+            vehicle.fields == {"BatteryLevel": {"interval_seconds": 60}}
+            and vehicle.preferTyped is True,
+            f"fields {vehicle.fields}, prefer_typed {vehicle.preferTyped}",
+        )
+    )
+
+    # Mutating the event dict after delivery must not corrupt the stored
+    # record - the config-sync listener also hands this same dict to public
+    # listeners, so storing an alias to it would let a consumer's in-place
+    # edit silently corrupt the last-known-good record.
+    aliasing_vehicle, aliasing_stream = make_vehicle()
+    assert aliasing_stream.config_listener is not None
+    delivered_event = {
+        "vin": VIN,
+        "config": {
+            "fields": {"CarType": {"interval_seconds": 60}},
+            "prefer_typed": False,
+        },
+    }
+    aliasing_stream.config_listener(delivered_event)
+    delivered_event["config"]["fields"]["CarType"]["interval_seconds"] = 999
+    delivered_event["config"]["fields"]["Injected"] = {"interval_seconds": 1}
+    results.append(
+        check(
+            "mutating the event dict after delivery does not corrupt the record",
+            aliasing_vehicle.fields == {"CarType": {"interval_seconds": 60}},
+            f"fields {aliasing_vehicle.fields}",
+        )
+    )
+
+    # A request that now matches the updated record is skipped - no PATCH sent.
+    await vehicle.add_field("BatteryLevel", 60)
+    results.append(
+        check(
+            "a request matching the updated record is skipped",
+            vehicle.sent == [],
+            f"sent {vehicle.sent}",
+        )
+    )
+
+    # A request that differs from the updated record is still sent.
+    await vehicle.add_field("BatteryLevel", 30)
+    results.append(
+        check(
+            "a request differing from the updated record is sent",
+            len(vehicle.sent) == 1
+            and vehicle.sent[0]["fields"]["BatteryLevel"] == {"interval_seconds": 30},
+            f"sent {vehicle.sent}",
+        )
+    )
+
+    # A malformed/partial config event never corrupts the record.
+    handler = CaptureWarnings()
+    logger = logging.getLogger("teslemetry_stream")
+    logger.addHandler(handler)
+    try:
+        vehicle, stream = make_vehicle()
+        assert stream.config_listener is not None
+        stream.config_listener(
+            {
+                "vin": VIN,
+                "config": {
+                    "fields": {"BatteryLevel": {}},
+                    "prefer_typed": False,
+                },
+            }
+        )
+        good_fields, good_typed = dict(vehicle.fields), vehicle.preferTyped
+
+        # A non-dict "config" body is entirely rejected and logged.
+        stream.config_listener({"vin": VIN, "config": "not-a-dict"})
+        results.append(
+            check(
+                "a non-dict config event is ignored, logged, and keeps last-good",
+                vehicle.fields == good_fields
+                and vehicle.preferTyped == good_typed
+                and any("malformed" in m.lower() for m in handler.messages),
+                f"fields {vehicle.fields}, warnings {handler.messages}",
+            )
+        )
+
+        # A partially malformed body applies the well-typed piece and keeps
+        # the last-good value for the malformed piece.
+        handler.messages.clear()
+        stream.config_listener(
+            {"vin": VIN, "config": {"fields": "not-a-dict", "prefer_typed": True}}
+        )
+        results.append(
+            check(
+                "a partial config event applies the good field, keeps the bad one",
+                vehicle.fields == good_fields
+                and vehicle.preferTyped is True
+                and any("malformed" in m.lower() for m in handler.messages),
+                f"fields {vehicle.fields}, prefer_typed {vehicle.preferTyped}, "
+                f"warnings {handler.messages}",
+            )
+        )
+
+        # A config event missing a key entirely leaves that piece untouched.
+        handler.messages.clear()
+        stream.config_listener(
+            {"vin": VIN, "config": {"fields": {"CarType": {}}}}
+        )
+        results.append(
+            check(
+                "a config event omitting prefer_typed leaves it unchanged",
+                vehicle.fields == {"CarType": {}} and vehicle.preferTyped is True,
+                f"fields {vehicle.fields}, prefer_typed {vehicle.preferTyped}",
+            )
+        )
+
+        # A field entry that isn't itself a dict (e.g. null) is malformed
+        # shape too, even though the outer "fields" value is a dict - the
+        # whole "fields" piece is rejected, not just the bad entry, so it
+        # can't leave a null in self.fields that later crashes add_field.
+        good_fields = dict(vehicle.fields)
+        handler.messages.clear()
+        stream.config_listener(
+            {
+                "vin": VIN,
+                "config": {"fields": {"CarType": {}, "BatteryLevel": None}},
+            }
+        )
+        results.append(
+            check(
+                "a null nested field entry rejects the whole fields piece",
+                vehicle.fields == good_fields
+                and any("malformed" in m.lower() for m in handler.messages),
+                f"fields {vehicle.fields}, warnings {handler.messages}",
+            )
+        )
+
+        # And, concretely, a later add_field for an unrelated field must not
+        # raise trying to .get() off the (rejected, never-stored) null entry.
+        try:
+            await vehicle.add_field("BatteryLevel", 60)
+            add_field_ok = True
+        except AttributeError as error:
+            add_field_ok = False
+            add_field_error = repr(error)
+        results.append(
+            check(
+                "add_field after a rejected null entry does not raise",
+                add_field_ok,
+                "" if add_field_ok else add_field_error,
+            )
+        )
+    finally:
+        logger.removeHandler(handler)
+
+    print("-" * 72)
+    print("ALL PASS" if all(results) else "FAILURES PRESENT")
+    if not all(results):
+        raise SystemExit(1)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/tests/test_config_listener_lifecycle.py b/tests/test_config_listener_lifecycle.py
new file mode 100644
index 0000000..f814754
--- /dev/null
+++ b/tests/test_config_listener_lifecycle.py
@@ -0,0 +1,248 @@
+"""Regression tests for the vehicle config-sync listener's lifecycle.
+
+Defects flagged across review of the config-consume feature:
+
+- A first attempt eagerly registered the internal config listener in
+  ``TeslemetryStreamVehicle.__init__``, which risked ``TeslemetryStream.
+  async_add_listener()`` -> ``asyncio.create_task()`` needing a running
+  event loop. That was worked around by deferring registration to first
+  use (inside ``add_field``/``prefer_typed``/``update_config``) - but
+  lazy registration left a gap: a stream that was already connected
+  before the listener existed could have dispatched a config event that
+  was simply never seen.
+- The actual fix for the loop hazard was structural, not timing:
+  ``async_add_listener(..., internal=True)`` makes its
+  ``schedule_refresh`` (the gate on the ``asyncio.create_task()`` call)
+  unconditionally false for an internal-only registration - so
+  registering the config listener eagerly in ``__init__`` is safe outside
+  a running loop, and the lazy-registration gap is gone: the listener
+  exists from construction, so no connection can ever predate it.
+- The same ``internal=True`` flag also excludes it from the "last
+  listener removed" auto-close check (and its counterpart "first listener
+  starts the task" check) - otherwise a permanently-registered internal
+  listener would keep ``_listeners`` non-empty forever, and a later public
+  listener's own zero-to-one transition couldn't restart a closed stream.
+- The config-sync listener can only observe a server-side change while
+  connected AND the ``config`` topic isn't filtered out via
+  ``TeslemetryStream(topics=...)``. A separate attempt at handling *that*
+  forced a REST refresh before the no-op check whenever disconnected -
+  reverted, since it added a failure path and could storm the API with
+  GETs for a batch of callers. The no-op *skip* is purely an optimization
+  (a redundant PATCH is harmless), so it's gated on the record actually
+  being live-maintained (``_record_is_live()``) rather than
+  force-freshened; otherwise add_field/prefer_typed just send
+  unconditionally, exactly the pre-feature status quo.
+"""
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+from teslemetry_stream.stream import TeslemetryStream
+from teslemetry_stream.vehicle import TeslemetryStreamVehicle
+
+VIN = "TESTVIN0000000001"
+
+
+class FakeSession:
+    """A session whose get() is never expected to be called in these tests."""
+
+    async def get(self, url: str, **kwargs: Any) -> Any:
+        raise AssertionError(f"unexpected session.get({url!r}) - these tests must not connect")
+
+
+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:
+    # manual=True: these tests exercise listener bookkeeping, not the real
+    # connect/listen loop - FakeSession.get() intentionally isn't a working
+    # SSE endpoint.
+    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
+
+
+def test_sync_construction_without_a_loop() -> bool:
+    """Must run before any event loop exists - constructing a stream+vehicle
+    synchronously (the library's documented pre-async-context usage) must not
+    require or start one, and the config listener must already be registered
+    by the time construction returns (internal=True never reaches the
+    asyncio.create_task() call that would need a loop)."""
+    label = "TeslemetryStream(vin=...) construction outside a running loop does not raise"
+    try:
+        # TeslemetryStream(vin=...) constructs its own TeslemetryStreamVehicle
+        # internally (get_vehicle), which is exactly the construction path
+        # that must stay loop-free.
+        stream = make_stream(vin=VIN)
+    except RuntimeError as error:
+        return check(label, False, f"raised {error!r}")
+    ok = check(label, True)
+    return check(
+        "the config listener is registered by the time construction returns",
+        len(stream._listeners) == 1,
+        f"listeners {len(stream._listeners)}",
+    ) and ok
+
+
+async def test_registration_happens_at_construction(results: list[bool]) -> None:
+    stream = make_stream()
+    _vehicle, _sent = make_vehicle_with_capture(stream)
+
+    results.append(
+        check(
+            "the internal config listener is registered by construction, before any call",
+            len(stream._listeners) == 1,
+            f"listeners {len(stream._listeners)}",
+        )
+    )
+    results.append(
+        check(
+            "the registered listener is marked internal",
+            all(is_internal for _, _, is_internal in stream._listeners.values()),
+        )
+    )
+
+
+async def test_auto_close_after_last_public_listener_removed(results: list[bool]) -> None:
+    stream = make_stream()
+    # The internal listener registers at construction; no call needed to set it up.
+    _vehicle, _sent = make_vehicle_with_capture(stream)
+
+    # A real/public listener on top of the internal one.
+    remove_public = stream.async_add_listener(lambda event: None)
+
+    results.append(
+        check(
+            "two listeners are registered: one internal, one public",
+            len(stream._listeners) == 2,
+            f"listeners {len(stream._listeners)}",
+        )
+    )
+
+    stream.active = True  # simulate a live connection to observe close() flip it back
+    remove_public()
+
+    results.append(
+        check(
+            "removing the last public listener still auto-closes the stream",
+            stream.active is False,
+        )
+    )
+    results.append(
+        check(
+            "the internal listener remains registered after auto-close",
+            len(stream._listeners) == 1,
+            f"listeners {len(stream._listeners)}",
+        )
+    )
+
+
+async def test_cold_stream_add_field_sends_patch_unconditionally(results: list[bool]) -> None:
+    """A never-connected (or disconnected) stream can't have observed a
+    server-side change, so the no-op skip must not apply - send the PATCH
+    unconditionally rather than trying to force the record fresh."""
+    stream = make_stream()
+    vehicle, sent = make_vehicle_with_capture(stream)
+    vehicle.fields = {"BatteryLevel": {"interval_seconds": 60}}  # matches the request below
+
+    results.append(check("the stream starts disconnected", not stream.connected))
+
+    await vehicle.add_field("BatteryLevel", 60)
+
+    results.append(
+        check(
+            "add_field sends the PATCH even though the record already matches",
+            len(sent) == 1 and sent[0]["fields"]["BatteryLevel"] == {"interval_seconds": 60},
+            f"sent {sent}",
+        )
+    )
+
+
+async def test_filtered_config_topic_sends_patch_unconditionally(results: list[bool]) -> None:
+    """Even while connected, if `topics=` filters out the config topic the
+    config-sync listener never receives anything - the record can't be
+    trusted, so the no-op skip must not apply."""
+    stream = make_stream(topics=["state"])
+    vehicle, sent = make_vehicle_with_capture(stream)
+    vehicle.fields = {"BatteryLevel": {"interval_seconds": 60}}
+
+    stream._response = object()  # type: ignore[assignment]  # simulate a live connection
+    results.append(check("the stream is connected", stream.connected))
+
+    await vehicle.add_field("BatteryLevel", 60)
+
+    results.append(
+        check(
+            "add_field sends the PATCH when the config topic is filtered out",
+            len(sent) == 1 and sent[0]["fields"]["BatteryLevel"] == {"interval_seconds": 60},
+            f"sent {sent}",
+        )
+    )
+
+
+async def test_connected_and_subscribed_record_match_skips(results: list[bool]) -> None:
+    """The no-op skip only applies once both conditions hold: connected, and
+    the config topic isn't filtered out (default `topics=None` subscribes
+    to everything)."""
+    stream = make_stream()
+    vehicle, sent = make_vehicle_with_capture(stream)
+    vehicle.fields = {"BatteryLevel": {"interval_seconds": 60}}
+
+    stream._response = object()  # type: ignore[assignment]  # simulate a live connection
+    results.append(
+        check(
+            "the stream is connected and subscribed to every topic",
+            stream.connected and stream.topics is None,
+        )
+    )
+
+    await vehicle.add_field("BatteryLevel", 60)
+
+    results.append(
+        check(
+            "add_field skips the PATCH when the record is live-maintained and matches",
+            sent == [],
+            f"sent {sent}",
+        )
+    )
+
+
+async def main(pre_loop_results: list[bool]) -> None:
+    results: list[bool] = list(pre_loop_results)
+    await test_registration_happens_at_construction(results)
+    await test_auto_close_after_last_public_listener_removed(results)
+    await test_cold_stream_add_field_sends_patch_unconditionally(results)
+    await test_filtered_config_topic_sends_patch_unconditionally(results)
+    await test_connected_and_subscribed_record_match_skips(results)
+
+    print("-" * 72)
+    print("ALL PASS" if all(results) else "FAILURES PRESENT")
+    if not all(results):
+        raise SystemExit(1)
+
+
+if __name__ == "__main__":
+    # Must run before asyncio.run() starts a loop - that's the entire point.
+    pre_loop_results = [test_sync_construction_without_a_loop()]
+    asyncio.run(main(pre_loop_results))
diff --git a/tests/test_config_update.py b/tests/test_config_update.py
index 62a3320..e06c73a 100644
--- a/tests/test_config_update.py
+++ b/tests/test_config_update.py
@@ -29,6 +29,15 @@ 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 that check 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 make_vehicle(vin: str, responses: list[Any]) -> TeslemetryStreamVehicle:
diff --git a/tests/test_energysite_events.py b/tests/test_energysite_events.py
index 9c8749e..f5bd827 100644
--- a/tests/test_energysite_events.py
+++ b/tests/test_energysite_events.py
@@ -108,7 +108,7 @@ def make_stream() -> TeslemetryStream:
 
 def dispatch(stream: TeslemetryStream, event: dict[str, Any]) -> None:
     """Replicate stream.listen()'s per-event dispatch without a live connection."""
-    for listener, filters in list(stream._listeners.values()):
+    for listener, filters, _internal in list(stream._listeners.values()):
         if recursive_match(filters, event):
             listener(event)
 
diff --git a/tests/test_field_type_coercion.py b/tests/test_field_type_coercion.py
index 42921f2..0c1f8e2 100644
--- a/tests/test_field_type_coercion.py
+++ b/tests/test_field_type_coercion.py
@@ -21,6 +21,10 @@ class FakeStream:
     """Minimal stand-in for TeslemetryStream that just captures listeners."""
 
     manual = True
+    # Keeps the record "live" (see _record_is_live) so add_field's no-op
+    # check short-circuits, matching this test's pre-populated fields.
+    connected = True
+    topics = None
 
     def __init__(self) -> None:
         # maps Signal value -> wrapped listener callback
@@ -30,11 +34,15 @@ def async_add_listener(
         self,
         callback: Callable[[dict[str, Any]], None],
         filters: dict[str, Any] | None = None,
+        internal: bool = False,
     ) -> Callable[[], None]:
-        # filters carries {"vin": ..., "data": {Signal: None}} — grab the field
+        # filters carries {"vin": ..., "data": {Signal: None}} — grab the field.
+        # The vehicle's own internal config-sync listener has no "data" key;
+        # it's not under test here, so just ignore it.
         assert filters is not None
-        signal = next(iter(filters["data"]))
-        self.captured[signal] = callback
+        if "data" in filters:
+            signal = next(iter(filters["data"]))
+            self.captured[signal] = callback
         return lambda: None
 
 
diff --git a/tests/test_sse_topics.py b/tests/test_sse_topics.py
index 1d898da..af1b430 100644
--- a/tests/test_sse_topics.py
+++ b/tests/test_sse_topics.py
@@ -79,7 +79,7 @@ def make_stream(topics: Any = None) -> tuple[TeslemetryStream, FakeSession]:
 
 def dispatch(stream: TeslemetryStream, event: dict[str, Any]) -> None:
     """Replicate stream.listen()'s per-event dispatch without a live connection."""
-    for listener, filters in list(stream._listeners.values()):
+    for listener, filters, _internal in list(stream._listeners.values()):
         if recursive_match(filters, event):
             listener(event)
 
diff --git a/tests/test_stream_lifecycle.py b/tests/test_stream_lifecycle.py
index 2b0cf47..c99875d 100644
--- a/tests/test_stream_lifecycle.py
+++ b/tests/test_stream_lifecycle.py
@@ -12,11 +12,15 @@
 
 import asyncio
 import contextlib
+from collections.abc import Callable
 from typing import Any
 
 import aiohttp
 
 from teslemetry_stream.stream import TeslemetryStream
+from teslemetry_stream.vehicle import TeslemetryStreamVehicle
+
+VIN = "TESTVIN0000000001"
 
 
 class FakeContent:
@@ -37,13 +41,35 @@ def fail(self, exc: BaseException) -> None:
             self._blocker.set_exception(exc)
 
 
+class FakeEventContent:
+    """Async-iterable response body yielding canned SSE `data:` lines, then
+    blocking until failed or cancelled (like `FakeContent`)."""
+
+    def __init__(self, lines: list[bytes]) -> None:
+        self._lines = list(lines)
+        self._blocker: asyncio.Future[None] = asyncio.get_running_loop().create_future()
+
+    def __aiter__(self) -> FakeEventContent:
+        return self
+
+    async def __anext__(self) -> bytes:
+        if self._lines:
+            return self._lines.pop(0)
+        await self._blocker
+        raise AssertionError("unreachable - blocker only resolves via an exception")
+
+    def fail(self, exc: BaseException) -> None:
+        if not self._blocker.done():
+            self._blocker.set_exception(exc)
+
+
 class FakeResponse:
     """Minimal stand-in for the aiohttp response `connect()` awaits."""
 
-    def __init__(self) -> None:
+    def __init__(self, content: Any = None) -> None:
         self.url = "https://fake.teslemetry.com/sse"
         self.status = 200
-        self.content = FakeContent()
+        self.content = content if content is not None else FakeContent()
         self.closed = False
 
     def close(self) -> None:
@@ -57,12 +83,16 @@ def __init__(self) -> None:
         self.calls = 0
         self.responses: list[FakeResponse] = []
         self.gate: asyncio.Event | None = None
+        # Overridable factory for the response's `content` - defaults to the
+        # blocks-forever FakeContent when unset.
+        self.content_factory: Callable[[], Any] | None = None
 
     async def get(self, url: str, **kwargs: Any) -> FakeResponse:
         self.calls += 1
         if self.gate is not None:
             await self.gate.wait()
-        response = FakeResponse()
+        content = self.content_factory() if self.content_factory else None
+        response = FakeResponse(content)
         self.responses.append(response)
         return response
 
@@ -243,6 +273,167 @@ async def test_close_prevents_reconnect_after_backoff(results: list[bool]) -> No
     )
 
 
+async def test_restart_after_public_readd_with_internal_listener_present(
+    results: list[bool],
+) -> None:
+    """An internal (bookkeeping-only) listener surviving auto-close must not
+    block a later public listener from restarting the owned task."""
+    session = FakeSession()
+    stream = make_stream(session)
+
+    # An internal listener alone must not itself start the task - only
+    # public listeners drive connect/disconnect.
+    remove_internal = stream.async_add_listener(lambda event: None, internal=True)
+    results.append(
+        check(
+            "an internal-only listener does not start the owned task",
+            stream._listen_task is None,
+        )
+    )
+
+    remove_public = stream.async_add_listener(lambda event: None)
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+    results.append(
+        check(
+            "adding the first public listener connects",
+            session.calls == 1,
+            f"got {session.calls}",
+        )
+    )
+
+    # Removing the last public listener auto-closes even though the
+    # internal listener remains registered.
+    remove_public()
+    await asyncio.sleep(0)
+    results.append(
+        check(
+            "removing the last public listener auto-closes despite the internal listener",
+            not stream.active,
+        )
+    )
+
+    # A later public listener, added while only the internal one remains
+    # registered, must still restart the owned task - this is the bug: the
+    # registry was non-empty (internal listener) so the old whole-registry
+    # emptiness check never saw a zero-to-one transition.
+    remove_public2 = stream.async_add_listener(lambda event: None)
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+    results.append(
+        check(
+            "re-adding a public listener afterwards reconnects",
+            session.calls == 2,
+            f"got {session.calls}",
+        )
+    )
+
+    remove_public2()
+    remove_internal()
+    stream.close()
+    await asyncio.sleep(0)
+
+
+async def test_dispatch_survives_listener_creating_vehicle_mid_iteration(
+    results: list[bool],
+) -> None:
+    """A callback that calls get_vehicle() for an uncached VIN - or otherwise
+    adds a listener - mid-dispatch inserts into `_listeners` while `listen()`
+    is iterating it. Dispatching over a snapshot means that must not raise
+    and kill the loop; both queued events should still be delivered."""
+    session = FakeSession()
+    session.content_factory = lambda: FakeEventContent(
+        [
+            b'data: {"vin": "A", "state": "online"}\n',
+            b'data: {"vin": "A", "state": "online"}\n',
+        ]
+    )
+    stream = make_stream(session)
+
+    delivered: list[dict[str, Any]] = []
+
+    def mutate_during_dispatch(event: dict[str, Any]) -> None:
+        delivered.append(event)
+        # Registers a new internal listener - a mid-dispatch mutation of
+        # the exact dict listen() is iterating.
+        stream.get_vehicle(f"NEWVIN{len(delivered)}")
+
+    stream.async_add_listener(mutate_during_dispatch, {"vin": None})
+
+    for _ in range(5):
+        await asyncio.sleep(0)
+
+    results.append(
+        check(
+            "the listen task survives a listener mutating _listeners mid-dispatch",
+            stream._listen_task is not None and not stream._listen_task.done(),
+        )
+    )
+    results.append(
+        check(
+            "both queued events are delivered - dispatch continues past the mutation",
+            len(delivered) == 2,
+            f"delivered {len(delivered)}",
+        )
+    )
+    results.append(
+        check(
+            "each callback-created vehicle registered its own internal listener",
+            len(stream.vehicles) == 2,
+            f"vehicles {list(stream.vehicles)}",
+        )
+    )
+
+    stream.close()
+    await asyncio.sleep(0)
+
+
+async def test_internal_listener_sees_event_before_public_mutator(results: list[bool]) -> None:
+    """A public listener registered BEFORE the internal one - and running
+    first in registration order - must not get a chance to mutate the event
+    in place before the internal (bookkeeping) listener has cached from it.
+    Dispatch order must be internal-first, not registration-order."""
+    session = FakeSession()
+    session.content_factory = lambda: FakeEventContent(
+        [
+            (
+                b'data: {"vin": "'
+                + VIN.encode()
+                + b'", "config": {"fields": '
+                + b'{"BatteryLevel": {"interval_seconds": 60}}, '
+                + b'"prefer_typed": true}}\n'
+            )
+        ]
+    )
+    stream = make_stream(session)
+
+    def public_mutator(event: dict[str, Any]) -> None:
+        # A badly-behaved public consumer mutating its event argument.
+        event["config"]["fields"]["BatteryLevel"]["interval_seconds"] = 999
+        event["config"]["prefer_typed"] = False
+
+    # Registered first (and would run first under registration order) but
+    # is not internal - the internal config listener, registered second
+    # (via vehicle construction below), must still see the event first.
+    stream.async_add_listener(public_mutator)
+    vehicle = TeslemetryStreamVehicle(stream, VIN)
+
+    for _ in range(5):
+        await asyncio.sleep(0)
+
+    results.append(
+        check(
+            "the internal listener captured the pristine value, not the public mutation",
+            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)
@@ -250,6 +441,9 @@ async def main() -> None:
     await test_cancel_while_blocked_reading(results)
     await test_close_during_connect(results)
     await test_close_prevents_reconnect_after_backoff(results)
+    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)
 
     print("-" * 72)
     print("ALL PASS" if all(results) else "FAILURES PRESENT")