diff --git a/AGENTS.md b/AGENTS.md
index 8930b8a..9bc42d4 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -6,7 +6,7 @@ This file is the project's committed home for project-intrinsic agent knowledge:
 - CI runs on push/PR: `.github/workflows/ci.yml`. Uses `uv` (see `uv.lock`). Three jobs: `lint` (ruff + mypy), `test` (matrix), `build` (`uv build` + `twine check` + artifact upload). Ruff and twine, like mypy, aren't declared dev dependencies - CI installs them ephemerally via `uv run --with <tool> <tool> ...`, matching the existing mypy pattern.
 - `[tool.ruff]` in `pyproject.toml` selects `E, F, W, I, B, UP, ASYNC, SIM, RUF` and ignores `RUF006` - the stream's background `listen`/refresh tasks (`stream.py`, `vehicle.py`) are intentionally untracked fire-and-forget `asyncio.create_task` calls, not a lint oversight.
 - The `test` job's step fails outright (non-zero exit) if `tests/test_*.py` matches nothing, rather than skipping - do not reintroduce a silent-skip fallback there.
-- `tests/` files are plain scripts (`if __name__ == "__main__"`), not pytest-based - pytest would collect zero tests here. Run each directly, e.g. `uv run python tests/test_field_type_coercion.py`.
+- `tests/` files are plain scripts (`if __name__ == "__main__"`), not pytest-based - pytest would collect zero tests here. Run each directly, e.g. `uv run python tests/test_config_events.py`.
 - `pyproject.toml` has a `[tool.mypy]` config but no dev-dependency group declares mypy, so `uv sync` alone won't install it. CI installs it ephemerally via `uv run --with mypy mypy teslemetry_stream`.
 - `Signal` in `const.py` tracks <https://api.teslemetry.com/fields.json>; the config route rejects names it does not know with `fst_err_validation`. Fields the API has retired are not rejected - it accepts the request and names them in a top-level `ignoredFields` list - so the library can lag the published list without breaking.
 - Config responses are shaped inconsistently: success is flat, `{"updated_vehicles": n}` plus `ignoredFields` when some were dropped, while errors are wrapped, `{"response": null, "error": ...}`. Do not look for `updated_vehicles` under `response`; that lookup silently never matches.
@@ -16,9 +16,9 @@ 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.
-- `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.
+- `TeslemetryStreamVehicle` keeps `fields` 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}}` 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` piece replaces the record (every nested entry must itself be a dict - one bad entry, e.g. a null, rejects the whole 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 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. `preferTyped`/`prefer_typed()` (the per-vehicle opt-in that used to control this) no longer exists in this library - prefer_typed is enabled by default server-side now, so there is nothing left to toggle. It is still only a default, not a guarantee: some vehicles haven't picked it up and still stream string-encoded numeric/boolean values, so `make_int`/`make_float`/`make_bool` (`vehicle.py`) keep coercing a `str` payload to native `int`/`float`/`bool` rather than assuming every vehicle is typed. `tests/test_field_type_coercion.py` covers this coercion against real observed telemetry.
 - `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.
+- `add_field` gates its 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` outright with no network call. `_populated` is set by a successful `get_config()` (200 or 404 - both are an authoritative answer; 404 also clears `fields`, since no config existing is itself the authoritative state, not a fetch to ignore) 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. A failed populating fetch (`aiohttp.ClientError`/timeout, or a non-200/404 status via `raise_for_status()`) is not authoritative like a 404: `_ensure_populated()` catches it, logs, and leaves the vehicle unpopulated rather than letting it propagate - every `listen_*` method reaches `add_field()` through `_enable_field()`'s fire-and-forget `asyncio.create_task()`, where an uncaught exception would just silently abandon the field request instead of sending the PATCH.
 - `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.
 
diff --git a/pyproject.toml b/pyproject.toml
index 099a1c7..a29fcc6 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ requires = ["setuptools>=77.0"]
 
 [project]
 name = "teslemetry_stream"
-version = "0.10.1"
+version = "0.11.0"
 license = "Apache-2.0"
 description = "Teslemetry Streaming API library for Python"
 readme = "README.md"
diff --git a/teslemetry_stream/vehicle.py b/teslemetry_stream/vehicle.py
index 3bc1bcf..5daf84e 100644
--- a/teslemetry_stream/vehicle.py
+++ b/teslemetry_stream/vehicle.py
@@ -69,7 +69,6 @@ class TeslemetryStreamVehicle:
     """Handle streaming field updates."""
 
     fields: dict[str, dict[str, int]]
-    preferTyped: bool | None
     _config: dict[str, Any]
     _flight: asyncio.Task[None] | None
     _populated: bool
@@ -82,11 +81,10 @@ def __init__(self, stream: TeslemetryStream, vin: str):
         self.lock = asyncio.Lock()
         # Per-instance: class-level dicts would share pending config between vehicles
         self.fields = {}
-        self.preferTyped = None
         self._config = {}
-        # Whether fields/preferTyped reflect a real server answer (a push
-        # event or a REST fetch) rather than just their unset defaults -
-        # gates add_field/prefer_typed's lazy REST fetch below.
+        # Whether fields reflects a real server answer (a push event or a
+        # REST fetch) rather than just its unset default - gates add_field's
+        # lazy REST fetch below.
         self._populated = False
         # The single in-flight (or most recently completed) coalesced flush.
         # Callers that arrive while it is running merge into `_config` and
@@ -106,10 +104,10 @@ def __init__(self, stream: TeslemetryStream, vin: str):
             {Key.VIN: self.vin, Key.CONFIG: None},
             internal=True,
         )
-        # A disconnect can leave fields/preferTyped stale until the next
-        # connection's config snapshot arrives - unpopulate so a
-        # field-config call landing in that window awaits a fresh fetch
-        # instead of trusting the pre-disconnect record.
+        # A disconnect can leave fields stale until the next connection's
+        # config snapshot arrives - unpopulate so a field-config call
+        # landing in that window awaits a fresh fetch instead of trusting
+        # the pre-disconnect record.
         self.stream.async_add_connection_listener(self._on_connection_event)
 
     @property
@@ -117,7 +115,6 @@ def config(self) -> dict[str, Any]:
         """Return current configuration."""
         return {
             "fields": self.fields,
-            "prefer_typed": self.preferTyped,
         }
 
     async def get_config(self) -> None:
@@ -133,12 +130,15 @@ async def get_config(self) -> None:
             response = await req.json()
 
             self.fields = response.get("fields", {})
-            self.preferTyped = response.get("prefer_typed", False)
             self._populated = True
             return
         if req.status == 404:
             # No config exists for this vehicle yet - an authoritative
-            # answer (empty), not a missing one.
+            # answer (empty), not a missing one. Clear any fields left over
+            # from before a disconnect: without this, a config deleted
+            # elsewhere while offline would leave add_field() no-op'ing
+            # forever against a record the server no longer has.
+            self.fields = {}
             self._populated = True
             return
 
@@ -163,14 +163,23 @@ async def _ensure_populated(self) -> None:
         if flight is None or flight.done():
             flight = asyncio.ensure_future(self.get_config())
             self._populate_flight = flight
-        await asyncio.shield(flight)
+        try:
+            await asyncio.shield(flight)
+        except (aiohttp.ClientError, asyncio.TimeoutError) as error:
+            # Unlike a 404, a failed fetch isn't authoritative - stay
+            # unpopulated and let the caller proceed against whatever
+            # `fields` currently holds, rather than raising into a
+            # fire-and-forget `_enable_field()` task where nobody would
+            # ever see the exception or retry the stranded field request.
+            LOGGER.warning(
+                "Config fetch failed for %s, proceeding without it: %s", self.vin, error
+            )
 
     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.
+        A well-typed `fields` piece replaces the record; a malformed one is
+        logged and skipped so it can't corrupt the last-known-good record.
         """
         config = event.get(Key.CONFIG)
         if not isinstance(config, dict):
@@ -201,17 +210,6 @@ def _on_config_event(self, event: dict[str, Any]) -> None:
                     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.
 
@@ -275,10 +273,6 @@ async def _flush(self) -> None:
                 }
                 LOGGER.debug("Configured streaming fields %s", ", ".join(applied))
                 self.fields = {**self.fields, **applied}
-            prefer_typed = self._config.get("prefer_typed")
-            if isinstance(prefer_typed, bool):
-                LOGGER.debug("Configured streaming typed to %s", prefer_typed)
-                self.preferTyped = prefer_typed
             self._config.clear()
 
     async def _patch_with_bounded_retry(
@@ -346,13 +340,6 @@ async def add_field(self, field: Signal | str, interval: int | None = None) -> N
         value = {"interval_seconds": interval} if interval else None
         await self.update_config({"fields": {field: value}})
 
-    async def prefer_typed(self, prefer_typed: bool) -> None:
-        """Set prefer typed."""
-        await self._ensure_populated()
-        if self.preferTyped == prefer_typed:
-            return
-        await self.update_config({"prefer_typed": prefer_typed})
-
     def _enable_field(self, field: Signal) -> None:
         """Enable a field for streaming from a listener."""
         asyncio.create_task(self.add_field(field))
@@ -2974,7 +2961,9 @@ def make_int(
     def typer(event: dict[str, Any]) -> None:
         data = event["data"][signal]
         if isinstance(data, str):
-            # Handle invalid and None?
+            # Some vehicles still stream string-encoded values even with
+            # prefer_typed enabled by default - keep coercing rather than
+            # assuming every vehicle is typed.
             data = int(data)
         callback(data)
 
@@ -2989,7 +2978,9 @@ def make_float(
     def typer(event: dict[str, Any]) -> None:
         data = event["data"][signal]
         if isinstance(data, str):
-            # Handle invalid and None?
+            # Some vehicles still stream string-encoded values even with
+            # prefer_typed enabled by default - keep coercing rather than
+            # assuming every vehicle is typed.
             data = float(data)
         callback(data)
 
@@ -3004,7 +2995,9 @@ def make_bool(
     def typer(event: dict[str, Any]) -> None:
         data = event["data"][signal]
         if isinstance(data, str):
-            # Handle invalid and None?
+            # Some vehicles still stream string-encoded values even with
+            # prefer_typed enabled by default - keep coercing rather than
+            # assuming every vehicle is typed.
             data = data == "true"
         callback(data)
 
diff --git a/tests/test_config_events.py b/tests/test_config_events.py
index 69d3c6e..201074c 100644
--- a/tests/test_config_events.py
+++ b/tests/test_config_events.py
@@ -1,14 +1,13 @@
 """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
+shaped like ``{"vin": ..., "config": {"fields": {...}}}``, 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.
+running loop) so ``fields`` - and therefore the ``add_field`` no-op check -
+reflects current server truth rather than only what this client has itself
+requested or observed at connect.
 """
 from __future__ import annotations
 
@@ -86,18 +85,14 @@ async def main() -> None:
     stream.config_listener(
         {
             "vin": VIN,
-            "config": {
-                "fields": {"BatteryLevel": {"interval_seconds": 60}},
-                "prefer_typed": True,
-            },
+            "config": {"fields": {"BatteryLevel": {"interval_seconds": 60}}},
         }
     )
     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}",
+            "a config event updates fields",
+            vehicle.fields == {"BatteryLevel": {"interval_seconds": 60}},
+            f"fields {vehicle.fields}",
         )
     )
 
@@ -109,10 +104,7 @@ async def main() -> None:
     assert aliasing_stream.config_listener is not None
     delivered_event = {
         "vin": VIN,
-        "config": {
-            "fields": {"CarType": {"interval_seconds": 60}},
-            "prefer_typed": False,
-        },
+        "config": {"fields": {"CarType": {"interval_seconds": 60}}},
     }
     aliasing_stream.config_listener(delivered_event)
     delivered_event["config"]["fields"]["CarType"]["interval_seconds"] = 999
@@ -154,15 +146,9 @@ async def main() -> None:
         vehicle, stream = make_vehicle()
         assert stream.config_listener is not None
         stream.config_listener(
-            {
-                "vin": VIN,
-                "config": {
-                    "fields": {"BatteryLevel": {}},
-                    "prefer_typed": False,
-                },
-            }
+            {"vin": VIN, "config": {"fields": {"BatteryLevel": {}}}}
         )
-        good_fields, good_typed = dict(vehicle.fields), vehicle.preferTyped
+        good_fields = dict(vehicle.fields)
 
         # A non-dict "config" body is entirely rejected and logged.
         stream.config_listener({"vin": VIN, "config": "not-a-dict"})
@@ -170,39 +156,31 @@ async def main() -> None:
             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.
+        # A malformed "fields" piece is rejected and logged, keeping last-good.
         handler.messages.clear()
-        stream.config_listener(
-            {"vin": VIN, "config": {"fields": "not-a-dict", "prefer_typed": True}}
-        )
+        stream.config_listener({"vin": VIN, "config": {"fields": "not-a-dict"}})
         results.append(
             check(
-                "a partial config event applies the good field, keeps the bad one",
+                "a malformed fields piece is rejected, keeps last-good",
                 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}",
+                f"fields {vehicle.fields}, warnings {handler.messages}",
             )
         )
 
-        # A config event missing a key entirely leaves that piece untouched.
+        # A config event omitting "fields" entirely leaves it unchanged.
         handler.messages.clear()
-        stream.config_listener(
-            {"vin": VIN, "config": {"fields": {"CarType": {}}}}
-        )
+        stream.config_listener({"vin": VIN, "config": {}})
         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 config event omitting fields leaves it unchanged",
+                vehicle.fields == good_fields,
+                f"fields {vehicle.fields}",
             )
         )
 
diff --git a/tests/test_config_listener_lifecycle.py b/tests/test_config_listener_lifecycle.py
index 21cb43e..b2debe7 100644
--- a/tests/test_config_listener_lifecycle.py
+++ b/tests/test_config_listener_lifecycle.py
@@ -6,8 +6,8 @@
   ``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
+  use (inside ``add_field``/``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:
@@ -22,18 +22,20 @@
   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.
-- ``add_field``/``prefer_typed`` no longer gate their no-op skip on
-  connection/topic state (``_record_is_live()``, since removed). Instead
-  they gate on whether the record has ever been ``_populated`` - by a
-  push event or by a REST fetch. An unpopulated record is fetched over
-  REST before the no-op decision is made; a populated one is trusted
-  without hitting the network at all.
+- ``add_field`` no longer gates its no-op skip on connection/topic state
+  (``_record_is_live()``, since removed). Instead it gates on whether the
+  record has ever been ``_populated`` - by a push event or by a REST
+  fetch. An unpopulated record is fetched over REST before the no-op
+  decision is made; a populated one is trusted without hitting the
+  network at all.
 """
 from __future__ import annotations
 
 import asyncio
 from typing import Any
 
+import aiohttp
+
 from teslemetry_stream.const import Key
 from teslemetry_stream.stream import TeslemetryStream
 from teslemetry_stream.vehicle import TeslemetryStreamVehicle
@@ -58,6 +60,20 @@ def __init__(self, status: int, body: dict[str, Any]) -> None:
     async def json(self) -> dict[str, Any]:
         return self._body
 
+    def raise_for_status(self) -> None:
+        if self.status >= 400:
+            url = "https://fake/api/config"
+            raise aiohttp.ClientResponseError(
+                request_info=aiohttp.RequestInfo(
+                    url=url,  # type: ignore[arg-type]
+                    method="GET",
+                    headers={},  # type: ignore[arg-type]
+                    real_url=url,  # type: ignore[arg-type]
+                ),
+                history=(),
+                status=self.status,
+            )
+
 
 class FetchingSession:
     """A session that serves a canned config GET and counts calls."""
@@ -72,6 +88,17 @@ async def get(self, url: str, **kwargs: Any) -> FakeConfigResponse:
         return FakeConfigResponse(self.status, self.body)
 
 
+class TransportFailingSession:
+    """A session whose config GET always raises a transport-level error."""
+
+    def __init__(self) -> None:
+        self.calls = 0
+
+    async def get(self, url: str, **kwargs: Any) -> Any:
+        self.calls += 1
+        raise aiohttp.ClientConnectionError("simulated connection failure")
+
+
 def check(label: str, ok: bool, detail: str = "") -> bool:
     print(f"{label:<72} {'PASS' if ok else 'FAIL'}{'  ' + detail if detail else ''}")
     return ok
@@ -201,7 +228,7 @@ async def test_unpopulated_add_field_fetches_before_deciding(results: list[bool]
     not trust its unset defaults - add_field awaits get_config() first, then
     decides the no-op skip against the answer it got back."""
     session = FetchingSession(
-        200, {"fields": {"BatteryLevel": {"interval_seconds": 60}}, "prefer_typed": False}
+        200, {"fields": {"BatteryLevel": {"interval_seconds": 60}}}
     )
     stream = make_stream(session)
     vehicle, sent = make_vehicle_with_capture(stream)
@@ -245,10 +272,7 @@ async def test_populated_add_field_skips_without_fetching(results: list[bool]) -
     vehicle._on_config_event(
         {
             Key.VIN: VIN,
-            Key.CONFIG: {
-                "fields": {"BatteryLevel": {"interval_seconds": 60}},
-                "prefer_typed": False,
-            },
+            Key.CONFIG: {"fields": {"BatteryLevel": {"interval_seconds": 60}}},
         }
     )
     results.append(check("the config event populated the vehicle", vehicle._populated))
@@ -263,12 +287,100 @@ async def test_populated_add_field_skips_without_fetching(results: list[bool]) -
     )
 
 
+async def test_authoritative_404_clears_stale_fields(results: list[bool]) -> None:
+    """A vehicle carrying fields from before a disconnect must not trust them
+    forever if the config was deleted server-side while it was offline - a
+    404 on the reconnect-window fetch is authoritative (no config exists),
+    so it must clear `fields`, not just mark the record populated."""
+    session = FetchingSession(404, {})
+    stream = make_stream(session)
+    vehicle, sent = make_vehicle_with_capture(stream)
+
+    # Simulate a stale record surviving a disconnect: previously populated
+    # with a field the server no longer has, then unpopulated as a real
+    # disconnect would leave it.
+    vehicle.fields = {"BatteryLevel": {"interval_seconds": 60}}
+    vehicle._populated = False
+
+    await vehicle.get_config()
+    results.append(
+        check(
+            "an authoritative 404 clears stale fields, not just marks populated",
+            vehicle.fields == {},
+            f"fields {vehicle.fields}",
+        )
+    )
+
+    # With the stale record cleared, a later add_field for that same field
+    # must send - not skip on a match that no longer exists - and without
+    # re-fetching, since the (now correctly empty) record is populated.
+    await vehicle.add_field("BatteryLevel", 60)
+    results.append(
+        check(
+            "add_field sends the PATCH instead of no-op'ing against the stale record",
+            session.calls == 1 and len(sent) == 1 and "BatteryLevel" in sent[0]["fields"],
+            f"calls {session.calls}, sent {sent}",
+        )
+    )
+
+
+async def test_unpopulated_add_field_survives_a_failed_status_fetch(
+    results: list[bool],
+) -> None:
+    """A non-200/404 status on the populating fetch (e.g. a transient 5xx)
+    is not authoritative like a 404 - it must not strand the field request
+    the way it silently would inside a fire-and-forget `_enable_field()`
+    task, where nobody would ever see the raised exception or retry.
+    add_field must stay unpopulated and still send the PATCH."""
+    session = FetchingSession(500, {})
+    stream = make_stream(session)
+    vehicle, sent = make_vehicle_with_capture(stream)
+
+    await vehicle.add_field("BatteryLevel", 60)
+    results.append(
+        check(
+            "add_field sends the PATCH despite the failed fetch, not stranding it",
+            len(sent) == 1 and "BatteryLevel" in sent[0]["fields"],
+            f"sent {sent}",
+        )
+    )
+    results.append(
+        check(
+            "the vehicle stays unpopulated after a failed (non-authoritative) fetch",
+            not vehicle._populated,
+        )
+    )
+
+
+async def test_unpopulated_add_field_survives_a_transport_failure(
+    results: list[bool],
+) -> None:
+    """A transport-level failure (ClientError/timeout) on the populating
+    fetch must be handled the same way as a bad status: proceed to the
+    PATCH rather than stranding the field request."""
+    session = TransportFailingSession()
+    stream = make_stream(session)
+    vehicle, sent = make_vehicle_with_capture(stream)
+
+    await vehicle.add_field("BatteryLevel", 60)
+    results.append(
+        check(
+            "add_field sends the PATCH despite a transport failure",
+            len(sent) == 1 and "BatteryLevel" in sent[0]["fields"],
+            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_unpopulated_add_field_fetches_before_deciding(results)
     await test_populated_add_field_skips_without_fetching(results)
+    await test_authoritative_404_clears_stale_fields(results)
+    await test_unpopulated_add_field_survives_a_failed_status_fetch(results)
+    await test_unpopulated_add_field_survives_a_transport_failure(results)
 
     print("-" * 72)
     print("ALL PASS" if all(results) else "FAILURES PRESENT")
diff --git a/tests/test_config_update.py b/tests/test_config_update.py
index 1c3f38f..f44d7e0 100644
--- a/tests/test_config_update.py
+++ b/tests/test_config_update.py
@@ -191,17 +191,6 @@ async def main() -> None:
         )
     )
 
-    # prefer_typed=False must be recorded as False, not coerced to True.
-    vehicle = make_vehicle(VIN_A, [ACCEPTED])
-    await vehicle.prefer_typed(False)
-    results.append(
-        check(
-            "prefer_typed(False) recorded as False",
-            vehicle.preferTyped is False,
-            f"got {vehicle.preferTyped!r}",
-        )
-    )
-
     # Pending config must not be shared between vehicles.
     first = make_vehicle(VIN_A, [FAILED])
     second = make_vehicle(VIN_B, [ACCEPTED])
diff --git a/tests/test_reconnect_config_window.py b/tests/test_reconnect_config_window.py
index 7760268..77fe400 100644
--- a/tests/test_reconnect_config_window.py
+++ b/tests/test_reconnect_config_window.py
@@ -113,7 +113,6 @@ async def test_reconnect_window_add_field_refetches_instead_of_trusting_stale_re
     # client yet) - the reconnect's config snapshot will reveal 30s.
     stream._session.config_body = {  # type: ignore[attr-defined]
         "fields": {"BatteryLevel": {"interval_seconds": 30}},
-        "prefer_typed": False,
     }
 
     # A connection listener reacting to reconnect - e.g. an HA integration
diff --git a/tests/test_stream_lifecycle.py b/tests/test_stream_lifecycle.py
index 146f898..e7e7f66 100644
--- a/tests/test_stream_lifecycle.py
+++ b/tests/test_stream_lifecycle.py
@@ -400,8 +400,7 @@ async def test_internal_listener_sees_event_before_public_mutator(results: list[
                 b'data: {"vin": "'
                 + VIN.encode()
                 + b'", "config": {"fields": '
-                + b'{"BatteryLevel": {"interval_seconds": 60}}, '
-                + b'"prefer_typed": true}}\n'
+                + b'{"BatteryLevel": {"interval_seconds": 60}}}}\n'
             )
         ]
     )
@@ -410,7 +409,6 @@ async def test_internal_listener_sees_event_before_public_mutator(results: list[
     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
@@ -424,9 +422,8 @@ def public_mutator(event: dict[str, Any]) -> None:
     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}",
+            vehicle.fields.get("BatteryLevel") == {"interval_seconds": 60},
+            f"fields {vehicle.fields}",
         )
     )
 
@@ -452,8 +449,7 @@ async def test_vehicle_discovered_mid_dispatch_fetches_correctly_on_first_use(
                 b'data: {"vin": "'
                 + new_vin.encode()
                 + b'", "config": {"fields": '
-                + b'{"BatteryLevel": {"interval_seconds": 60}}, '
-                + b'"prefer_typed": true}}\n'
+                + b'{"BatteryLevel": {"interval_seconds": 60}}}}\n'
             )
         ]
     )
@@ -493,7 +489,6 @@ async def get_config() -> None:
         # Stand in for the REST fetch a real session would serve - reflects
         # what the missed event actually carried.
         vehicle.fields = {"BatteryLevel": {"interval_seconds": 60}}
-        vehicle.preferTyped = True
         vehicle._populated = True
 
     vehicle.get_config = get_config  # type: ignore[assignment,method-assign]
diff --git a/uv.lock b/uv.lock
index 1884a2e..1d72232 100644
--- a/uv.lock
+++ b/uv.lock
@@ -612,7 +612,7 @@ wheels = [
 
 [[package]]
 name = "teslemetry-stream"
-version = "0.10.1"
+version = "0.11.0"
 source = { editable = "." }
 dependencies = [
     { name = "aiohttp" },