refactor: adopt cleaner write pattern for options arrays

fix(fridge): switch turn-off silently sends 'On' (truthy-string guard)

custom_components/localthings/config_flow.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import datetime
+import json
 import logging
 import re
 import selectors
@@ -15,6 +16,10 @@ from homeassistant import config_entries
 from homeassistant.core import callback
 from homeassistant.data_entry_flow import FlowResult
 from homeassistant.helpers.selector import (
+    ObjectSelector,
+    SelectSelector,
+    SelectSelectorConfig,
+    SelectSelectorMode,
     TextSelector,
     TextSelectorConfig,
     TextSelectorType,
@@ -396,22 +401,44 @@ class LocalThingsConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
 
 
 class LocalThingsOptionsFlow(config_entries.OptionsFlow):
-    """Per-device override of the remote-control-off write block (issue
-    #54). The block exists because most devices reject writes outright
-    while remote control is off and a clear error beats a silent
+    """Per-device options: the remote-control-off write-block override
+    (issue #54) plus a debug panel for writing an arbitrary body to an
+    arbitrary resource href, so a user can pin down device-specific write
+    behavior without waiting on a new release.
+
+    The remote-control override exists because most devices reject writes
+    outright while remote control is off and a clear error beats a silent
     device-side rejection -- but not every model actually enforces that,
     so this lets a user who's confirmed their device accepts writes anyway
     turn the block off for just that device rather than it being
-    hardcoded on for everyone."""
+    hardcoded on for everyone. The debug panel goes further: it bypasses
+    that block (and every write_fn/validate_fn) entirely, sending exactly
+    the body the user types to whatever href they pick.
+    """
+
+    def __init__(self) -> None:
+        self._debug_href: str = ""
+        self._debug_result: tuple[int, dict] | None = None
+
+    def _coordinator(self):
+        return self.hass.data.get(DOMAIN, {}).get(self.config_entry.entry_id)
 
     async def async_step_init(
         self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        return self.async_show_menu(
+            step_id="init",
+            menu_options=["settings", "debug_write"],
+        )
+
+    async def async_step_settings(
+        self, user_input: dict[str, Any] | None = None
     ) -> FlowResult:
         if user_input is not None:
             return self.async_create_entry(data=user_input)
 
         return self.async_show_form(
-            step_id="init",
+            step_id="settings",
             data_schema=vol.Schema({
                 vol.Required(
                     CONF_BYPASS_REMOTE_CONTROL,
@@ -421,3 +448,96 @@ class LocalThingsOptionsFlow(config_entries.OptionsFlow):
                 ): bool,
             }),
         )
+
+    async def async_step_debug_write(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        coord = self._coordinator()
+        if coord is None:
+            return self.async_abort(reason="not_loaded")
+
+        if user_input is not None:
+            self._debug_href = user_input["href"]
+            return await self.async_step_debug_edit()
+
+        hrefs = sorted(coord.last_resources.keys())
+        return self.async_show_form(
+            step_id="debug_write",
+            data_schema=vol.Schema({
+                vol.Required("href"): SelectSelector(SelectSelectorConfig(
+                    options=hrefs,
+                    custom_value=True,
+                    mode=SelectSelectorMode.DROPDOWN,
+                )),
+            }),
+        )
+
+    def _show_debug_edit_form(
+        self, href: str, current: dict, errors: dict[str, str], payload,
+    ) -> FlowResult:
+        return self.async_show_form(
+            step_id="debug_edit",
+            data_schema=vol.Schema({
+                vol.Required(
+                    "payload", default=(payload if payload is not None else {}),
+                ): ObjectSelector(),
+            }),
+            errors=errors,
+            description_placeholders={
+                "href": href,
+                "current_value": (
+                    json.dumps(current, indent=2, ensure_ascii=False)
+                    if current else "(no cached value)"
+                ),
+            },
+        )
+
+    async def async_step_debug_edit(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        coord = self._coordinator()
+        if coord is None:
+            return self.async_abort(reason="not_loaded")
+
+        href = self._debug_href
+        current = coord.resource(href)
+
+        if user_input is not None:
+            payload = user_input.get("payload") or {}
+            if not isinstance(payload, dict) or not payload:
+                return self._show_debug_edit_form(
+                    href, current, {"payload": "empty_payload"}, payload
+                )
+            try:
+                code, new_rep = await coord.async_raw_write(href, payload)
+            except Exception:  # noqa: BLE001 - surfaced to the user below
+                _LOGGER.exception("debug raw write failed for %s", href)
+                return self._show_debug_edit_form(
+                    href, current, {"base": "write_failed"}, payload
+                )
+            self._debug_result = (code, new_rep)
+            return await self.async_step_debug_result()
+
+        return self._show_debug_edit_form(href, current, {}, None)
+
+    async def async_step_debug_result(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        code, new_rep = self._debug_result or (0, {})
+        return self.async_show_menu(
+            step_id="debug_result",
+            menu_options=["debug_write", "finish"],
+            description_placeholders={
+                "code": f"{code >> 5}.{code & 0x1f:02d} ({code:#04x})",
+                "new_value": (
+                    json.dumps(new_rep, indent=2, ensure_ascii=False)
+                    if new_rep else "(no value returned)"
+                ),
+            },
+        )
+
+    async def async_step_finish(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        # Close the flow without altering saved options.
+        return self.async_create_entry(data=dict(self.config_entry.options))

custom_components/localthings/coordinator.py

@@ -22,7 +22,7 @@ from smartthings_local.ocf.state_cache import StateCache
 
 from .registry.batch import parse_device0_batch
 from .registry.by_type import for_device, for_device_by_model, for_device_by_resources
-from .registry.capabilities.common import remote_control_enabled
+from .registry.capabilities.common import merge_options_field, remote_control_enabled
 from .registry.discovery import discover, BoundEntity
 from .registry import CAPABILITIES
 from .registry.adapter import flatten
@@ -612,7 +612,24 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         # introduced its own races around overlapping writes to the same
         # href. Simpler and safer to just hold the guard for the full,
         # generously-sized window and let it expire on its own.
-        self._observe.apply(write_href, body, source='optimistic')
+        # write_fn bodies that touch x.com.samsung.da.options carry only the
+        # changed token(s) now (issue #54: confirmed sufficient on the wire --
+        # the device merges by prefix itself), not the whole packed array.
+        # observe.apply()'s field-level {**cached, **rep} merge doesn't know
+        # that -- handed the bare token list, it would replace the cached
+        # field outright and wipe every sibling option for the rest of the
+        # settle window. Pre-merge it here the same way the device does, so
+        # the optimistic cache entry stays complete; the minimal `body` below
+        # is still exactly what goes out over the wire.
+        optimistic_body = body
+        new_options = body.get('x.com.samsung.da.options')
+        if isinstance(new_options, list):
+            cached_options = (self._cache.get(write_href) or {}).get('x.com.samsung.da.options')
+            optimistic_body = {
+                **body,
+                'x.com.samsung.da.options': merge_options_field(cached_options, new_options),
+            }
+        self._observe.apply(write_href, optimistic_body, source='optimistic')
         self._observe.mark_write_pending(
             write_href, settle_s=self._POST_TIMEOUT_S + self._POLL_TIMEOUT_S
         )
@@ -630,3 +647,58 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
             self._log.error("command failed for %s: %s", write_href, e)
         else:
             await self.async_request_refresh()
+
+    # ------------------------------------------------------------------
+    # Debug raw write (issue #54): a power-user escape hatch for the
+    # options-flow debug panel, letting a user POST an arbitrary partial
+    # body to an arbitrary href to pin down device-specific write behavior
+    # without waiting on a new release. Deliberately bypasses the
+    # remote-control block and every write_fn/validate_fn above -- that's
+    # the whole point, so use with care.
+    # ------------------------------------------------------------------
+
+    def _raw_write_blocking(self, path_segs: list[str], body: dict, href: str) -> tuple[int, dict]:
+        """Debug primitive: POST an arbitrary patch, then read the href
+        back for ground truth. Blocking -- runs in executor."""
+        if self._session is None:
+            self._connect_session()
+        sess = self._session
+        if sess is None:
+            raise RuntimeError("no session")
+        code, _ = sess.post(path_segs, cbor2.dumps(body), timeout=self._POST_TIMEOUT_S)
+        self._log.warning("DEBUG raw write POST %s %r → code %#04x", href, body, code)
+        new_rep: dict = {}
+        try:
+            sess.pace()
+            rcode, payload = sess.get(path_segs, timeout=10.0)
+            if rcode == 0x45 and payload:
+                rep = cbor2.loads(payload)
+                if isinstance(rep, dict):
+                    self._observe.apply(href, rep, source='poll')
+                    new_rep = rep
+        except Exception as e:
+            self._log.debug("raw write follow-up read failed: %s", e)
+        return code, new_rep
+
+    async def async_raw_write(self, href: str, body: dict) -> tuple[int, dict]:
+        """Debug-only arbitrary write (issue #54). Bypasses the
+        remote-control block and all write_fn/validate_fn logic; sends
+        `body` verbatim as a partial-rep PATCH to `href`. Returns
+        (coap_code, new_rep) where new_rep is the href's value read back
+        right after the write. Used by the options-flow debug panel to
+        help users pin down device-specific write behavior without a new
+        release."""
+        if not isinstance(body, dict) or not body:
+            raise ServiceValidationError("Debug write payload must be a non-empty object.")
+        path_segs = [s for s in href.strip('/').split('/') if s]
+        if not path_segs:
+            raise ServiceValidationError("A resource href is required.")
+        norm_href = '/' + '/'.join(path_segs)
+        async with self._session_lock:
+            code, new_rep = await self.hass.async_add_executor_job(
+                self._raw_write_blocking, path_segs, body, norm_href
+            )
+        # Hasten a full summary poll so entities on other resources catch
+        # up too -- a debug write can affect siblings, not just its href.
+        await self.async_request_refresh()
+        return code, new_rep

custom_components/localthings/registry/capabilities/air_purifier.py

@@ -1,6 +1,5 @@
 """Capabilities for the Samsung ARTIK051_TVTL-class air purifier family
-(model AX60R5080WD/SE, issue #56 -- verified against two independent
-diagnostics dumps of the same internal model).
+(model AX60R5080WD/SE, issue #56).
 
 Power, kids-lock, remote-control, alarms, and the energy meter are the shared
 common.py capabilities (this family exposes the standard /power/0+/power/vs/0
@@ -8,37 +7,44 @@ pair and /alarms/vs/0, /energy/consumption/vs/0). /diagnosis/vs/0 reuses
 dishwasher.DIAGNOSIS -- identical field/write contract
 (x.com.samsung.da.diagnosisStart, 'Ready' on both dumps).
 
-Two things are deliberately left as raw, unwritable diagnostic sensors rather
-than modeled as real controls, per the "don't guess" rule:
+/mode/vs/0's x.com.samsung.da.options array packs multiple independent
+'<Prefix>_<value>' flags into one list -- the same packed-list contract
+laundry.py's option_value/option_write already model for /course/vs/0's
+options[] (reused directly below, just against this family's own href). Per
+issue #56's follow-up (five diagnostics dumps captured with the physical unit
+set to Auto/Sleep/Low/Medium/High):
+  Light_On / Light_Off  -- a plain on/off flag; MODE below models it as a
+                            real switch, RMW-replacing just that one entry.
+  Comode_Off            -- read 'Off' on *every* one of the five dumps,
+                            including High/Low/Medium/Auto -- confirms this
+                            is NOT the fan-speed selector (ruling out the
+                            original guess); exposed read-only since its
+                            actual purpose is still unconfirmed.
+  OptionCode_60282       -- confirmed opaque/not user-facing in the
+                            SmartThings app; not modeled (same treatment as
+                            range_hood's OptionCode_* token on the same
+                            href).
+  Blooming_*             -- confirmed to have no corresponding SmartThings
+                            app setting; dropped entirely rather than kept
+                            as an unexplained diagnostic (it did track 1:1
+                            with Sleep mode across the five dumps -- 0 in
+                            Sleep, 6 otherwise -- so it's plausibly an
+                            automatic side effect of sleep mode, e.g. a
+                            display-dimming level, but that's still a guess).
 
-  /airflow/0, /airflow/vs/0 -- OCF-standard + vendor pair for fan speed/
-    direction, both zeroed/'Off' on every dump seen (device was off in both).
-    No supportedSpeed/supportedModes list is present anywhere in either dump
-    to confirm the valid range, so a write-capable fan/select isn't safe to
-    build yet -- see the issue #56 request for a running-state dump.
-
-  /mode/vs/0's x.com.samsung.da.options array packs multiple independent
-    '<Prefix>_<value>' flags into one list -- the same packed-list/RMW
-    contract laundry.py's option_value/replace_in_options already model for
-    /course/vs/0's options[] (reused directly below, just against this
-    family's own href). Of the tokens seen:
-      Light_On / Light_Off       -- read as a plain on/off flag; MODE below
-                                     models it as a real switch, RMW-
-                                     replacing just that one list entry.
-      Comode_Off                 -- never seen non-'Off' on these dumps;
-                                     likely the fan operating mode the issue
-                                     describes (Auto/Sleep/1/2/3), but
-                                     unconfirmed -- exposed read-only.
-      Blooming_0 / Blooming_6     -- meaning unconfirmed; exposed read-only.
-      OptionCode_60282            -- opaque, unchanged across both dumps;
-                                     not modeled (same treatment as
-                                     range_hood's OptionCode_* token on the
-                                     same href).
+/airflow/0 and /airflow/vs/0's `speed` still isn't modeled as a real
+fan-speed control: across the same five dumps it read 0 for both Auto *and*
+High, and 3 for Low/Medium *and* Sleep -- not a monotonic mapping to any
+selectable level, and the dumps were all captured within about three minutes
+of each other (only one poll cycle apart at this integration's 30s summary
+interval), so the values may not have settled after each change before the
+diagnostics snapshot was taken. Exposed read-only pending a confirmed,
+stable capture -- see the issue #56 discussion for what's needed.
 """
 from ..capability import Capability
 from ..entities import BinarySensorDesc, SensorDesc, SwitchDesc
 from .common import int_or_none, sensor_item_value
-from .laundry import bool_option_exists, bool_option_value, option_value, replace_in_options
+from .laundry import bool_option_exists, bool_option_value, option_value, option_write
 
 _AIR_QUALITY_SENSORS = (
     ('dust', 'Dust', 'mdi:blur', 'Dust'),
@@ -68,10 +74,12 @@ def _consumable_state(items, name):
     return None
 
 
-# FilterProgress is a raw 0-100 percentage in both dumps (100 and 62); which
-# end of that scale means "replace me" isn't confirmed from the dump alone,
-# so the entity is named after the raw field rather than asserting a
-# direction (see issue #56 follow-up questions).
+# FilterProgress is a 0-100 percentage counting up as the filter wears --
+# confirmed via issue #56: the SmartThings app shows "Filter needs changing"
+# once this reaches 100, so 100 means fully used, not "brand new." Named
+# after the raw field (matching the AC/range_hood filterUsage convention,
+# which counts the same direction) rather than "filter life," which would
+# imply the opposite direction.
 FILTER = Capability(
     href='/consumable/vs/0',
     poll_tier='cold',
@@ -127,9 +135,14 @@ AIRFLOW_VS_FALLBACK = Capability(
 
 
 def _light_write(payload, rep, href=None):
-    opts = list(rep.get('x.com.samsung.da.options') or [])
+    # option_write's single-token write is confirmed on a washer's
+    # /course/vs/0 (issue #54), NOT independently on this family's
+    # /mode/vs/0 -- extrapolated on the assumption the same vendor field
+    # merges the same way everywhere. If some unit replaces the field
+    # outright instead, this would drop Comode/OptionCode alongside it on
+    # the next light toggle; revisit if a real device report surfaces that.
     return ['mode', 'vs', '0'], {
-        'x.com.samsung.da.options': replace_in_options(opts, 'Light', payload),
+        'x.com.samsung.da.options': option_write('Light', payload),
     }
 
 
@@ -142,15 +155,12 @@ MODE = Capability(
                    rep_fn=bool_option_value('Light'),
                    exists_fn=bool_option_exists('Light'),
                    write_fn=_light_write),
-        # Read-only pending issue #56 follow-up -- see module docstring.
+        # Read-only -- confirmed NOT the fan-speed selector (see module
+        # docstring), actual purpose still unconfirmed.
         SensorDesc(key='operating_mode', name='Operating mode', icon='mdi:fan',
                    entity_category='diagnostic',
                    rep_fn=lambda rep: option_value(rep.get('x.com.samsung.da.options'), 'Comode'),
                    exists_fn=bool_option_exists('Comode')),
-        SensorDesc(key='blooming_level', name='Blooming level', icon='mdi:flower',
-                   entity_category='diagnostic',
-                   rep_fn=lambda rep: option_value(rep.get('x.com.samsung.da.options'), 'Blooming'),
-                   exists_fn=bool_option_exists('Blooming')),
     ),
 )
 

custom_components/localthings/registry/capabilities/common.py

@@ -65,6 +65,33 @@ def _active_alarm_codes(items):
     return ', '.join(codes) if codes else 'none'
 
 
+def merge_options_field(cached, new_tokens):
+    """Merge freshly-written `<Prefix>_<Value>` tokens into a cached
+    x.com.samsung.da.options[]-style array the same way the device itself
+    merges them: match by prefix, replace if present, append if not.
+
+    Confirmed on real hardware (issue #54) that a write only needs to carry
+    the changed token(s), not the whole array -- see laundry.option_write /
+    oven._option_write for the write side. This is the read side of that
+    same fact: coordinator.async_send_command uses it to keep the
+    optimistic cache entry for the written href complete (every sibling
+    option still present) during the write-settle window, since the wire
+    body it applies straight to the cache no longer carries them."""
+    merged = list(cached or [])
+    for token in new_tokens or ():
+        if not isinstance(token, str) or '_' not in token:
+            continue
+        prefix = token.split('_', 1)[0]
+        replaced = False
+        for i, o in enumerate(merged):
+            if isinstance(o, str) and o.startswith(prefix + '_'):
+                merged[i] = token
+                replaced = True
+        if not replaced:
+            merged.append(token)
+    return merged
+
+
 def sensor_item_value(items, sensor_type, index=0):
     """Pull one reading out of a `/sensors/vs/0`-style items[] list -- each
     item is `{type, value: [...]}`; `index` picks which slot of a possibly

custom_components/localthings/registry/capabilities/fridge.py

@@ -92,7 +92,7 @@ ICEMAKER_NIGHTTIME = Capability(
                    value_fn=lambda v: v == 'On',
                    write_fn=lambda p, rep, href=None: (
                        ['icemaker', 'nighttime', 'vs', '0'],
-                       {'ice.night.status': 'On' if p else 'Off'})),
+                       {'ice.night.status': 'On' if p == 'On' else 'Off'})),
     ),
 )
 
@@ -168,7 +168,7 @@ DOOR_ALERT = Capability(
 def _status_lock_write(field):
     return lambda p, rep, href=None: (
         ['status', 'lock', 'vs', '0'],
-        {field: 'On' if p else 'Off'}
+        {field: 'On' if p == 'On' else 'Off'}
     )
 
 
@@ -212,7 +212,7 @@ DEFROST_DELAY = Capability(
                    value_fn=lambda v: v == 'On',
                    write_fn=lambda p, rep, href=None: (
                        ['defrost', 'delay', 'vs', '0'],
-                       {'x.com.samsung.da.delayDefrost': 'On' if p else 'Off'})),
+                       {'x.com.samsung.da.delayDefrost': 'On' if p == 'On' else 'Off'})),
     ),
 )
 
@@ -300,7 +300,7 @@ WELCOME_LIGHTING = Capability(
                    value_fn=lambda v: v == 'On',
                    write_fn=lambda p, rep, href=None: (
                        ['proximity', 'vs', '0'],
-                       {'status': 'On' if p else 'Off'})),
+                       {'status': 'On' if p == 'On' else 'Off'})),
     ),
 )
 
@@ -440,7 +440,7 @@ CABINET_LIGHT = Capability(
                    value_fn=lambda v: v == 'On',
                    write_fn=lambda p, rep, href=None: (
                        ['cabinet', 'light', 'total', 'vs', '0'],
-                       {'light.dimming.status': 'On' if p else 'Off'})),
+                       {'light.dimming.status': 'On' if p == 'On' else 'Off'})),
     ),
 )
 
@@ -669,7 +669,7 @@ ICEMAKER_STATUS_FALLBACK = Capability(
                    value_fn=lambda v: v == 'On',
                    write_fn=lambda p, rep, href=None: (
                        ['icemaker', 'status', 'vs', '0'],
-                       {'x.com.samsung.da.iceMaker': 'On' if p else 'Off'})),
+                       {'x.com.samsung.da.iceMaker': 'On' if p == 'On' else 'Off'})),
     ),
 )
 

custom_components/localthings/registry/capabilities/laundry.py

@@ -146,9 +146,13 @@ BUZZER_SOUND = Capability(
 # Cycle selection over /course/vs/0.
 #
 # The selected course and every other user-tunable option ride in the
-# x.com.samsung.da.options array on /course/vs/0 as `<Prefix>_<value>` tokens;
-# a write is a read-modify-write of that whole array (cycle_write). The set of
-# *selectable* courses is not hardcoded -- it's read live from
+# x.com.samsung.da.options array on /course/vs/0 as `<Prefix>_<value>` tokens.
+# Confirmed on real hardware (issue #54): a write only needs to carry the one
+# changed token -- `{'x.com.samsung.da.options': ['SoftenerLevelCtrl_2']}` --
+# the device matches by prefix, evicts the stale token, and merges the result
+# into the array itself. No read-modify-write of the whole array needed (see
+# option_write). The set of *selectable* courses is not hardcoded -- it's read
+# live from
 # x.com.samsung.da.editCourseList on /wm/editcourse/vs/0 (cycle_options), so we
 # never show a course a given model doesn't have or hide one it does. Course
 # codes are uppercase hex; display names live in translations under
@@ -257,17 +261,17 @@ def _course_codes_from_supported_options(course_rep):
     return []
 
 
-def replace_in_options(options, prefix, new_value):
-    return [f"{prefix}_{new_value}" if isinstance(o, str) and o.startswith(prefix + '_') else o
-            for o in options]
+def option_write(prefix, new_value):
+    """A one-token x.com.samsung.da.options write -- see the module comment
+    above cycle_options for why this doesn't read/rewrite the whole array."""
+    return [f'{prefix}_{new_value}']
 
 
 def cycle_write(p, rep, href=None):
-    opts = list(rep.get('x.com.samsung.da.options') or [])
-    if not opts:
+    if not rep.get('x.com.samsung.da.options'):
         return None
     return ['course', 'vs', '0'], {
-        'x.com.samsung.da.options': replace_in_options(opts, 'Course', p),
+        'x.com.samsung.da.options': option_write('Course', p),
     }
 
 
@@ -339,11 +343,10 @@ def bool_option_write(prefix):
     def write(p, rep, href=None):
         if p not in ('On', 'Off'):
             return None
-        opts = list(rep.get('x.com.samsung.da.options') or [])
-        if not opts:
+        if not rep.get('x.com.samsung.da.options'):
             return None
         return ['course', 'vs', '0'], {
-            'x.com.samsung.da.options': replace_in_options(opts, prefix, p),
+            'x.com.samsung.da.options': option_write(prefix, p),
         }
     return write
 

custom_components/localthings/registry/capabilities/oven.py

@@ -122,10 +122,17 @@ def _option_value(options, prefix):
     return None
 
 
-def _replace_in_options(options, prefix, new_value):
-    """Return a new options list with the `<prefix>_*` slot replaced."""
-    return [f"{prefix}_{new_value}" if o.startswith(prefix + '_') else o
-            for o in options]
+def _option_write(prefix, new_value):
+    """A one-token x.com.samsung.da.options write, mirroring
+    laundry.option_write. NOT independently confirmed on an oven -- issue
+    #54 only confirmed prefix-merge-on-write for a washer's /course/vs/0.
+    This extrapolates that same vendor field/contract to the oven's
+    /mode/vs/0, on the assumption the firmware handles the array the same
+    way there. If that assumption is wrong for some oven, a device that
+    replaces the field outright instead of merging would drop every other
+    option in it (Sound/fastpreheat/etc.) on the next write -- revisit if a
+    real device report surfaces that."""
+    return [f'{prefix}_{new_value}']
 
 
 # ---------------------------------------------------------------------------
@@ -175,47 +182,41 @@ def _oven_mode_write(p, rep, href=None):
 def _lamp_write(p, rep, href=None):
     if p not in ('On', 'Off'):
         return None
-    opts = list(rep.get('x.com.samsung.da.options') or [])
-    if not opts:
+    if not rep.get('x.com.samsung.da.options'):
         return None
     return ['mode', 'vs', '0'], {
-        'x.com.samsung.da.options': _replace_in_options(opts, 'UpperLamp', p),
+        'x.com.samsung.da.options': _option_write('UpperLamp', p),
     }
 
 
 def _sound_write(p, rep, href=None):
     if p not in ('On', 'Off'):
         return None
-    opts = list(rep.get('x.com.samsung.da.options') or [])
-    if not opts:
+    if not rep.get('x.com.samsung.da.options'):
         return None
     return ['mode', 'vs', '0'], {
-        'x.com.samsung.da.options': _replace_in_options(opts, 'Sound', p),
+        'x.com.samsung.da.options': _option_write('Sound', p),
     }
 
 
 def _fastpreheat_write(p, rep, href=None):
     if p not in ('On', 'Off'):
         return None
-    opts = list(rep.get('x.com.samsung.da.options') or [])
-    if not opts:
+    if not rep.get('x.com.samsung.da.options'):
         return None
     return ['mode', 'vs', '0'], {
-        'x.com.samsung.da.options': _replace_in_options(opts, 'fastpreheat', p),
+        'x.com.samsung.da.options': _option_write('fastpreheat', p),
     }
 
 
 def _naturalsteam_write(p, rep, href=None):
     if p not in ('On', 'Off'):
         return None
-    opts = list(rep.get('x.com.samsung.da.options') or [])
-    if not opts:
+    if not rep.get('x.com.samsung.da.options'):
         return None
-    if not any(o.startswith('NaturalSteam_') for o in opts):
-        opts = opts + [f'NaturalSteam_{p}']
-    else:
-        opts = _replace_in_options(opts, 'NaturalSteam', p)
-    return ['mode', 'vs', '0'], {'x.com.samsung.da.options': opts}
+    return ['mode', 'vs', '0'], {
+        'x.com.samsung.da.options': _option_write('NaturalSteam', p),
+    }
 
 
 # ---------------------------------------------------------------------------

custom_components/localthings/registry/capabilities/washer.py

@@ -19,7 +19,7 @@ from ..capability import Capability
 from ..entities import BinarySensorDesc, SelectDesc, SensorDesc
 from .laundry import (
     bool_option_exists, bool_option_switch, cycle_options, cycle_select, hex_pairs, option_value,
-    replace_in_options,
+    option_write,
 )
 
 # ---------------------------------------------------------------------------
@@ -215,8 +215,7 @@ def _dosing_level(prefix):
 
 def _level_write(prefix):
     def write(p, rep, href=None):
-        opts = list(rep.get('x.com.samsung.da.options') or [])
-        if not opts:
+        if not rep.get('x.com.samsung.da.options'):
             return None
         # `p` is the zero-padded supported code the UI selected (e.g. '03');
         # the device stores the level un-padded (e.g. '3'), matching how it
@@ -226,7 +225,7 @@ def _level_write(prefix):
         except (TypeError, ValueError):
             native = p
         return ['course', 'vs', '0'], {
-            'x.com.samsung.da.options': replace_in_options(opts, prefix, native),
+            'x.com.samsung.da.options': option_write(prefix, native),
         }
     return write
 

custom_components/localthings/strings.json

@@ -250,11 +250,47 @@
     "step": {
       "init": {
         "title": "Local Things Options",
+        "menu_options": {
+          "settings": "Remote control write settings",
+          "debug_write": "Debug: write to a resource"
+        }
+      },
+      "settings": {
+        "title": "Remote control write settings",
         "description": "Some devices accept certain writes (e.g. default detergent/softener dosing on a washer) even while reporting remote control off. By default, LocalThings blocks every write with a clear error whenever a device reports remote control off, rather than letting the device silently reject it. Only enable this if you've confirmed writes actually work on this device with remote control off -- otherwise you'll trade that clear error for a silent failure.",
         "data": {
           "bypass_remote_control_lock": "Allow writes even when remote control is reported off"
         }
+      },
+      "debug_write": {
+        "title": "Debug: write to a resource",
+        "description": "Power-user tool for pinning down device-specific write behavior. Pick the resource (href) you want to write to, or type a custom one that isn't listed. This bypasses the remote-control-off block and sends exactly the fields you provide -- it can misconfigure your appliance, so use it deliberately.",
+        "data": {
+          "href": "Resource href"
+        }
+      },
+      "debug_edit": {
+        "title": "Debug write: {href}",
+        "description": "Current value of `{href}`:\n```\n{current_value}\n```\nEnter ONLY the field(s) you want to change below. What you enter is sent to the device as-is (a partial update); it is NOT merged with the fields shown above, so don't paste the whole value back. Mind the types: a numeric string like \"1\" must stay quoted -- a bare 1 is sent as an integer, which some devices reject.",
+        "data": {
+          "payload": "Payload to write"
+        }
+      },
+      "debug_result": {
+        "title": "Debug write result",
+        "description": "The device returned CoAP code {code}. A 2.xx class means the write was accepted; 4.xx/5.xx means it was rejected. The resource now reads:\n```\n{new_value}\n```\nIf the value is unchanged, the device likely dropped the write. Pick what to do next:",
+        "menu_options": {
+          "debug_write": "Write another resource",
+          "finish": "Finish"
+        }
       }
+    },
+    "error": {
+      "empty_payload": "Enter at least one field to write.",
+      "write_failed": "The write failed. Check the Home Assistant logs for details."
+    },
+    "abort": {
+      "not_loaded": "This device isn't connected yet. Try again once it has loaded."
     }
   },
   "issues": {

custom_components/localthings/translations/en.json

@@ -250,11 +250,47 @@
     "step": {
       "init": {
         "title": "Local Things Options",
+        "menu_options": {
+          "settings": "Remote control write settings",
+          "debug_write": "Debug: write to a resource"
+        }
+      },
+      "settings": {
+        "title": "Remote control write settings",
         "description": "Some devices accept certain writes (e.g. default detergent/softener dosing on a washer) even while reporting remote control off. By default, LocalThings blocks every write with a clear error whenever a device reports remote control off, rather than letting the device silently reject it. Only enable this if you've confirmed writes actually work on this device with remote control off -- otherwise you'll trade that clear error for a silent failure.",
         "data": {
           "bypass_remote_control_lock": "Allow writes even when remote control is reported off"
         }
+      },
+      "debug_write": {
+        "title": "Debug: write to a resource",
+        "description": "Power-user tool for pinning down device-specific write behavior. Pick the resource (href) you want to write to, or type a custom one that isn't listed. This bypasses the remote-control-off block and sends exactly the fields you provide -- it can misconfigure your appliance, so use it deliberately.",
+        "data": {
+          "href": "Resource href"
+        }
+      },
+      "debug_edit": {
+        "title": "Debug write: {href}",
+        "description": "Current value of `{href}`:\n```\n{current_value}\n```\nEnter ONLY the field(s) you want to change below. What you enter is sent to the device as-is (a partial update); it is NOT merged with the fields shown above, so don't paste the whole value back. Mind the types: a numeric string like \"1\" must stay quoted -- a bare 1 is sent as an integer, which some devices reject.",
+        "data": {
+          "payload": "Payload to write"
+        }
+      },
+      "debug_result": {
+        "title": "Debug write result",
+        "description": "The device returned CoAP code {code}. A 2.xx class means the write was accepted; 4.xx/5.xx means it was rejected. The resource now reads:\n```\n{new_value}\n```\nIf the value is unchanged, the device likely dropped the write. Pick what to do next:",
+        "menu_options": {
+          "debug_write": "Write another resource",
+          "finish": "Finish"
+        }
       }
+    },
+    "error": {
+      "empty_payload": "Enter at least one field to write.",
+      "write_failed": "The write failed. Check the Home Assistant logs for details."
+    },
+    "abort": {
+      "not_loaded": "This device isn't connected yet. Try again once it has loaded."
     }
   },
   "issues": {

tests/fixtures/golden/air_purifier.json

@@ -1,7 +1,6 @@
 {
   "state_keys": [
     "alarm_code",
-    "blooming_level",
     "clean_level",
     "device_active",
     "diagnosis_status",

tests/localthings/test_config_flow.py

@@ -279,6 +279,20 @@ def test_probe_marks_washer_as_recognized(monkeypatch):
     assert recognized is True
 
 
+async def test_options_flow_init_shows_menu(hass: HomeAssistant) -> None:
+    """The options flow's entry point is now a menu (issue #54's debug
+    panel lives alongside the remote-control settings), not the settings
+    form directly."""
+    entry = MockConfigEntry(domain=DOMAIN, data=ENTRY_DATA, unique_id=f'localthings_{MOCK_SERIAL}')
+    entry.add_to_hass(hass)
+
+    result = await hass.config_entries.options.async_init(entry.entry_id)
+
+    assert result['type'] == FlowResultType.MENU
+    assert result['step_id'] == 'init'
+    assert set(result['menu_options']) == {'settings', 'debug_write'}
+
+
 async def test_options_flow_default_is_off(hass: HomeAssistant) -> None:
     """The bypass defaults to False, so devices that never touch this
     option see no change in the remote-control write block."""
@@ -286,9 +300,12 @@ async def test_options_flow_default_is_off(hass: HomeAssistant) -> None:
     entry.add_to_hass(hass)
 
     result = await hass.config_entries.options.async_init(entry.entry_id)
+    result = await hass.config_entries.options.async_configure(
+        result['flow_id'], user_input={'next_step_id': 'settings'}
+    )
 
     assert result['type'] == FlowResultType.FORM
-    assert result['step_id'] == 'init'
+    assert result['step_id'] == 'settings'
     assert result['data_schema']({})[CONF_BYPASS_REMOTE_CONTROL] is False
 
 
@@ -299,6 +316,9 @@ async def test_options_flow_can_enable_bypass(hass: HomeAssistant) -> None:
     entry.add_to_hass(hass)
 
     result = await hass.config_entries.options.async_init(entry.entry_id)
+    result = await hass.config_entries.options.async_configure(
+        result['flow_id'], user_input={'next_step_id': 'settings'}
+    )
     result = await hass.config_entries.options.async_configure(
         result['flow_id'], user_input={CONF_BYPASS_REMOTE_CONTROL: True}
     )
@@ -317,5 +337,140 @@ async def test_options_flow_reflects_previously_saved_value(hass: HomeAssistant)
     entry.add_to_hass(hass)
 
     result = await hass.config_entries.options.async_init(entry.entry_id)
+    result = await hass.config_entries.options.async_configure(
+        result['flow_id'], user_input={'next_step_id': 'settings'}
+    )
 
     assert result['data_schema']({})[CONF_BYPASS_REMOTE_CONTROL] is True
+
+
+async def test_options_flow_debug_write_shows_hrefs_from_coordinator(
+    hass: HomeAssistant, mock_coordinator_session,
+) -> None:
+    """The debug panel's href dropdown is populated from the live
+    coordinator's cached resources, not a static list."""
+    entry = MockConfigEntry(domain=DOMAIN, data=ENTRY_DATA, unique_id=f'localthings_{MOCK_SERIAL}')
+    entry.add_to_hass(hass)
+    await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+
+    result = await hass.config_entries.options.async_init(entry.entry_id)
+    result = await hass.config_entries.options.async_configure(
+        result['flow_id'], user_input={'next_step_id': 'debug_write'}
+    )
+
+    assert result['type'] == FlowResultType.FORM
+    assert result['step_id'] == 'debug_write'
+
+
+async def test_options_flow_debug_write_aborts_when_device_not_loaded(
+    hass: HomeAssistant,
+) -> None:
+    """No coordinator in hass.data (device never finished loading) means
+    the debug panel has nothing to write to or read from."""
+    entry = MockConfigEntry(domain=DOMAIN, data=ENTRY_DATA, unique_id=f'localthings_{MOCK_SERIAL}')
+    entry.add_to_hass(hass)
+
+    result = await hass.config_entries.options.async_init(entry.entry_id)
+    result = await hass.config_entries.options.async_configure(
+        result['flow_id'], user_input={'next_step_id': 'debug_write'}
+    )
+
+    assert result['type'] == FlowResultType.ABORT
+    assert result['reason'] == 'not_loaded'
+
+
+async def test_options_flow_debug_edit_writes_and_shows_result(
+    hass: HomeAssistant, mock_coordinator_session,
+) -> None:
+    """Picking an href, then submitting a payload, drives
+    coordinator.async_raw_write and lands on the result menu with the
+    device's response."""
+    entry = MockConfigEntry(domain=DOMAIN, data=ENTRY_DATA, unique_id=f'localthings_{MOCK_SERIAL}')
+    entry.add_to_hass(hass)
+    await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+
+    result = await hass.config_entries.options.async_init(entry.entry_id)
+    result = await hass.config_entries.options.async_configure(
+        result['flow_id'], user_input={'next_step_id': 'debug_write'}
+    )
+    result = await hass.config_entries.options.async_configure(
+        result['flow_id'], user_input={'href': '/washer/vs/0'},
+    )
+    assert result['type'] == FlowResultType.FORM
+    assert result['step_id'] == 'debug_edit'
+
+    with patch(
+        'custom_components.localthings.coordinator.LocalThingsCoordinator.async_raw_write',
+        return_value=(0x44, {'a': 1}),
+    ):
+        result = await hass.config_entries.options.async_configure(
+            result['flow_id'], user_input={'payload': {'a': 1}},
+        )
+
+    assert result['type'] == FlowResultType.MENU
+    assert result['step_id'] == 'debug_result'
+    assert result['description_placeholders']['code'] == '2.04 (0x44)'
+
+
+async def test_options_flow_debug_edit_rejects_empty_payload(
+    hass: HomeAssistant, mock_coordinator_session,
+) -> None:
+    entry = MockConfigEntry(domain=DOMAIN, data=ENTRY_DATA, unique_id=f'localthings_{MOCK_SERIAL}')
+    entry.add_to_hass(hass)
+    await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+
+    result = await hass.config_entries.options.async_init(entry.entry_id)
+    result = await hass.config_entries.options.async_configure(
+        result['flow_id'], user_input={'next_step_id': 'debug_write'}
+    )
+    result = await hass.config_entries.options.async_configure(
+        result['flow_id'], user_input={'href': '/washer/vs/0'},
+    )
+    result = await hass.config_entries.options.async_configure(
+        result['flow_id'], user_input={'payload': {}},
+    )
+
+    assert result['type'] == FlowResultType.FORM
+    assert result['step_id'] == 'debug_edit'
+    assert result['errors'] == {'payload': 'empty_payload'}
+
+
+async def test_options_flow_finish_preserves_existing_options(
+    hass: HomeAssistant, mock_coordinator_session,
+) -> None:
+    """Finishing from the debug-result menu must not clobber a previously
+    saved remote-control bypass setting."""
+    entry = MockConfigEntry(
+        domain=DOMAIN, data=ENTRY_DATA, unique_id=f'localthings_{MOCK_SERIAL}',
+        options={CONF_BYPASS_REMOTE_CONTROL: True},
+    )
+    entry.add_to_hass(hass)
+    await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+
+    result = await hass.config_entries.options.async_init(entry.entry_id)
+    result = await hass.config_entries.options.async_configure(
+        result['flow_id'], user_input={'next_step_id': 'debug_write'}
+    )
+    result = await hass.config_entries.options.async_configure(
+        result['flow_id'], user_input={'href': '/washer/vs/0'},
+    )
+    with patch(
+        'custom_components.localthings.coordinator.LocalThingsCoordinator.async_raw_write',
+        return_value=(0x44, {'a': 1}),
+    ):
+        result = await hass.config_entries.options.async_configure(
+            result['flow_id'], user_input={'payload': {'a': 1}},
+        )
+    assert result['type'] == FlowResultType.MENU
+    assert result['step_id'] == 'debug_result'
+
+    result = await hass.config_entries.options.async_configure(
+        result['flow_id'], user_input={'next_step_id': 'finish'}
+    )
+
+    assert result['type'] == FlowResultType.CREATE_ENTRY
+    assert entry.options[CONF_BYPASS_REMOTE_CONTROL] is True

tests/test_air_purifier_capabilities.py

@@ -42,7 +42,7 @@ def test_expected_entities_present():
         'power_switch', 'alarm_code', 'dust', 'fine_dust', 'super_fine_dust',
         'odor', 'clean_level', 'filter_progress', 'device_active',
         'diagnosis_status', 'fan_speed_level', 'fan_direction',
-        'display_light', 'operating_mode', 'blooming_level',
+        'display_light', 'operating_mode',
     ):
         assert key in state, key
 
@@ -59,6 +59,8 @@ def test_air_quality_sensor_values():
 
 
 def test_filter_progress_reads_named_consumable_item():
+    """FilterProgress is confirmed (issue #56) to count up as the filter
+    wears -- 100 means fully used and needs replacing, not "brand new"."""
     assert _state()['filter_progress'] == 100
 
 
@@ -71,9 +73,10 @@ def test_diagnosis_reuses_dishwasher_capability():
 
 
 def test_light_switch_write_contract():
-    """The display-light switch RMW-replaces only the 'Light_*' entry in the
-    packed /mode/vs/0 options list (via laundry.replace_in_options), leaving
-    the other flags and the list order untouched."""
+    """The display-light switch writes only the changed 'Light_*' token
+    (via laundry.option_write) -- confirmed on real hardware (issue #54)
+    that the device merges by prefix itself, so no read-modify-write of the
+    whole packed /mode/vs/0 options list is needed."""
     desc = next(e for e in air_purifier.MODE.entities if e.key == 'display_light')
     rep = {'x.com.samsung.da.options': [
         'Comode_Off', 'Blooming_0', 'Light_On', 'OptionCode_60282',
@@ -81,24 +84,29 @@ def test_light_switch_write_contract():
     assert desc.rep_fn(rep) is True
     assert desc.write_fn('Off', rep) == (
         ['mode', 'vs', '0'],
-        {'x.com.samsung.da.options': [
-            'Comode_Off', 'Blooming_0', 'Light_Off', 'OptionCode_60282',
-        ]},
+        {'x.com.samsung.da.options': ['Light_Off']},
     )
 
 
-def test_mode_tokens_are_read_only_diagnostics():
-    """Comode_/Blooming_ tokens surface as raw diagnostic sensors rather than
-    a select/control -- their valid value ranges aren't confirmed yet (see
-    the air_purifier.py module docstring and the issue #56 follow-up)."""
+def test_operating_mode_is_a_read_only_diagnostic():
+    """Comode_* surfaces as a raw diagnostic sensor rather than a select/
+    control -- issue #56's five running-state dumps confirmed it reads 'Off'
+    regardless of the device's actual fan setting, ruling out the original
+    guess that it was the fan-speed selector; its real purpose is still
+    unconfirmed (see the air_purifier.py module docstring)."""
     operating_mode = next(e for e in air_purifier.MODE.entities if e.key == 'operating_mode')
-    blooming = next(e for e in air_purifier.MODE.entities if e.key == 'blooming_level')
-    rep = {'x.com.samsung.da.options': ['Comode_Off', 'Blooming_6']}
+    rep = {'x.com.samsung.da.options': ['Comode_Off']}
     assert operating_mode.rep_fn(rep) == 'Off'
-    assert blooming.rep_fn(rep) == '6'
     assert not hasattr(operating_mode, 'write_fn')
 
 
+def test_blooming_not_modeled():
+    """Confirmed (issue #56) to have no corresponding SmartThings app
+    setting -- dropped entirely rather than kept as an unexplained
+    diagnostic."""
+    assert not any(e.key == 'blooming_level' for e in air_purifier.MODE.entities)
+
+
 def test_airflow_vs_fallback_only_binds_without_generic():
     """/airflow/vs/0 is a match_fn fallback -- it must not bind when the
     OCF-standard /airflow/0 is also present (both are on every dump seen)."""

tests/test_common_capabilities.py

@@ -26,6 +26,37 @@ def test_common_caps_discover_on_dishwasher(dishwasher_resources):
     assert 'power_switch' in keys
 
 
+class TestMergeOptionsField:
+    """merge_options_field() is the read side of issue #54's finding: a
+    write only needs to carry the changed token, so the coordinator uses
+    this to keep its optimistic cache entry complete (every sibling option
+    still present) without waiting on a real poll."""
+
+    def test_replaces_matching_prefix(self):
+        cached = ['DeviceType_0167', 'Course_16', 'GMT_04']
+        assert common.merge_options_field(cached, ['Course_1D']) == [
+            'DeviceType_0167', 'Course_1D', 'GMT_04',
+        ]
+
+    def test_appends_when_prefix_absent(self):
+        cached = ['DeviceType_0167']
+        assert common.merge_options_field(cached, ['NaturalSteam_On']) == [
+            'DeviceType_0167', 'NaturalSteam_On',
+        ]
+
+    def test_merges_multiple_tokens_independently(self):
+        cached = ['DetergentLevelCtrl_1', 'SoftenerLevelCtrl_0', 'GMT_04']
+        merged = common.merge_options_field(cached, ['SoftenerLevelCtrl_2'])
+        assert merged == ['DetergentLevelCtrl_1', 'SoftenerLevelCtrl_2', 'GMT_04']
+
+    def test_handles_missing_cache(self):
+        assert common.merge_options_field(None, ['Course_1D']) == ['Course_1D']
+
+    def test_ignores_malformed_tokens(self):
+        cached = ['Course_16']
+        assert common.merge_options_field(cached, ['nounderscore']) == ['Course_16']
+
+
 # ---------------------------------------------------------------------------
 # OCF-native / vendor '-vs' fallback pairs (power, kids-lock, remote control).
 # ---------------------------------------------------------------------------

tests/test_coordinator_raw_write.py

@@ -0,0 +1,137 @@
+"""Tests for LocalThingsCoordinator.async_raw_write -- the debug-only
+arbitrary-href write primitive backing the options-flow debug panel
+(issue #54). It deliberately bypasses the remote-control block and every
+write_fn/validate_fn, so this only exercises the primitive itself: it
+POSTs exactly the caller's body, reads the href back, and validates its
+inputs.
+"""
+from __future__ import annotations
+
+from unittest.mock import AsyncMock
+
+import cbor2
+import pytest
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ServiceValidationError
+from pytest_homeassistant_custom_component.common import MockConfigEntry
+
+from custom_components.localthings.const import (
+    CONF_HOST, CONF_LEAF_CERT_PEM, CONF_LEAF_KEY_PEM, CONF_PORT, DOMAIN,
+)
+from custom_components.localthings.coordinator import LocalThingsCoordinator
+
+ENTRY_DATA = {
+    CONF_HOST: '10.0.0.199',
+    CONF_PORT: 49154,
+    CONF_LEAF_CERT_PEM: '-----BEGIN CERTIFICATE-----\nTEST-LEAF\n-----END CERTIFICATE-----',
+    CONF_LEAF_KEY_PEM: '-----BEGIN PRIVATE KEY-----\nTEST-LEAF-KEY\n-----END PRIVATE KEY-----',
+}
+
+
+class _FakeRawWriteSession:
+    """Stand-in for DtlsCoapSession: records every POST verbatim and
+    answers the follow-up GET with a canned representation -- no real
+    DTLS/network involved."""
+
+    def __init__(self, post_code: int = 0x44, get_rep: dict | None = None):
+        self.post_calls: list[tuple[list[str], bytes]] = []
+        self.get_calls: list[list[str]] = []
+        self._post_code = post_code
+        self._get_rep = {} if get_rep is None else get_rep
+
+    def post(self, path_segs, payload, timeout=None):
+        self.post_calls.append((list(path_segs), payload))
+        return self._post_code, b''
+
+    def get(self, path_segs, timeout=None):
+        self.get_calls.append(list(path_segs))
+        return 0x45, cbor2.dumps(self._get_rep)
+
+    def pace(self):
+        pass
+
+
+@pytest.fixture
+def coordinator(hass: HomeAssistant) -> LocalThingsCoordinator:
+    entry = MockConfigEntry(
+        domain=DOMAIN, data=ENTRY_DATA, unique_id='localthings_RAWWRITE-TEST',
+    )
+    entry.add_to_hass(hass)
+    coord = LocalThingsCoordinator(hass, entry)
+    # async_raw_write kicks a full refresh after the write; a real refresh
+    # would try to poll a session that doesn't exist for this unit test, so
+    # replace it with a no-op the same way the reference test setup does.
+    coord.async_request_refresh = AsyncMock()
+    return coord
+
+
+async def test_raw_write_splits_href_and_posts_exact_body(coordinator) -> None:
+    fake = _FakeRawWriteSession(post_code=0x44, get_rep={'x.field': 'after'})
+    coordinator._session = fake
+
+    body = {'x.com.samsung.da.field': 'value', 'n': 1}
+    code, new_rep = await coordinator.async_raw_write('/course/vs/0', body)
+
+    assert len(fake.post_calls) == 1
+    posted_path, posted_bytes = fake.post_calls[0]
+    assert posted_path == ['course', 'vs', '0']
+    assert cbor2.loads(posted_bytes) == body
+
+    assert code == 0x44
+    assert new_rep == {'x.field': 'after'}
+    coordinator.async_request_refresh.assert_awaited_once()
+
+
+async def test_raw_write_reads_href_back_for_ground_truth(coordinator) -> None:
+    fake = _FakeRawWriteSession(post_code=0x45, get_rep={'value': 'confirmed'})
+    coordinator._session = fake
+
+    code, new_rep = await coordinator.async_raw_write('/washer/vs/0', {'a': 1})
+
+    assert fake.get_calls == [['washer', 'vs', '0']]
+    assert code == 0x45
+    assert new_rep == {'value': 'confirmed'}
+
+
+async def test_raw_write_returns_coap_code_from_post_not_get(coordinator) -> None:
+    """The returned code reflects the write's own response, even though a
+    follow-up GET (which could carry a different code in principle) runs
+    right after it."""
+    fake = _FakeRawWriteSession(post_code=0x80, get_rep={'value': 1})
+    coordinator._session = fake
+
+    code, _ = await coordinator.async_raw_write('/test/vs/0', {'a': 1})
+
+    assert code == 0x80
+
+
+async def test_raw_write_rejects_empty_dict(coordinator) -> None:
+    with pytest.raises(ServiceValidationError):
+        await coordinator.async_raw_write('/washer/vs/0', {})
+
+
+async def test_raw_write_rejects_non_dict_payload(coordinator) -> None:
+    with pytest.raises(ServiceValidationError):
+        await coordinator.async_raw_write('/washer/vs/0', ['not', 'a', 'dict'])  # type: ignore[arg-type]
+
+
+async def test_raw_write_rejects_empty_href(coordinator) -> None:
+    with pytest.raises(ServiceValidationError):
+        await coordinator.async_raw_write('', {'a': 1})
+
+
+async def test_raw_write_rejects_root_href(coordinator) -> None:
+    with pytest.raises(ServiceValidationError):
+        await coordinator.async_raw_write('/', {'a': 1})
+
+
+async def test_raw_write_validation_errors_do_not_touch_the_session(coordinator) -> None:
+    """A rejected call must fail before ever reaching the network -- no
+    session needs to be connected at all for validation to run."""
+    assert coordinator._session is None
+
+    with pytest.raises(ServiceValidationError):
+        await coordinator.async_raw_write('/washer/vs/0', {})
+
+    assert coordinator._session is None
+    coordinator.async_request_refresh.assert_not_awaited()

tests/test_coordinator_send_command.py

@@ -0,0 +1,94 @@
+"""Tests for LocalThingsCoordinator.async_send_command's optimistic-apply
+step, specifically for x.com.samsung.da.options[] writes (issue #54).
+
+The write itself only needs to carry the single changed token -- confirmed
+on real hardware, the device merges by prefix and evicts the stale token
+itself. But observe.ObserveManager.apply() does a shallow {**cached, **rep}
+field merge, so handing it that same minimal single-token body would
+overwrite the *whole* cached options[] field, wiping every sibling option
+(other courses/levels/toggles packed into the same array) until the next
+real poll lands. async_send_command must pre-merge the token into the
+cached array itself before applying it optimistically, while still POSTing
+only the minimal body over the wire.
+"""
+from __future__ import annotations
+
+from unittest.mock import AsyncMock
+
+import cbor2
+import pytest
+from homeassistant.core import HomeAssistant
+from pytest_homeassistant_custom_component.common import MockConfigEntry
+
+from custom_components.localthings.const import (
+    CONF_HOST, CONF_LEAF_CERT_PEM, CONF_LEAF_KEY_PEM, CONF_PORT, DOMAIN,
+)
+from custom_components.localthings.coordinator import LocalThingsCoordinator
+from custom_components.localthings.registry.capabilities import laundry
+from custom_components.localthings.registry.discovery import BoundEntity
+
+ENTRY_DATA = {
+    CONF_HOST: '10.0.0.199',
+    CONF_PORT: 49154,
+    CONF_LEAF_CERT_PEM: '-----BEGIN CERTIFICATE-----\nTEST-LEAF\n-----END CERTIFICATE-----',
+    CONF_LEAF_KEY_PEM: '-----BEGIN PRIVATE KEY-----\nTEST-LEAF-KEY\n-----END PRIVATE KEY-----',
+}
+
+
+class _FakeSendSession:
+    def __init__(self):
+        self.post_calls: list[tuple[list[str], bytes]] = []
+
+    def post(self, path_segs, payload, timeout=None):
+        self.post_calls.append((list(path_segs), payload))
+        return 0x44, b''
+
+    def pace(self):
+        pass
+
+
+@pytest.fixture
+def coordinator(hass: HomeAssistant) -> LocalThingsCoordinator:
+    entry = MockConfigEntry(
+        domain=DOMAIN, data=ENTRY_DATA, unique_id='localthings_SENDCMD-TEST',
+    )
+    entry.add_to_hass(hass)
+    coord = LocalThingsCoordinator(hass, entry)
+    coord.async_request_refresh = AsyncMock()
+    coord._session = _FakeSendSession()
+    return coord
+
+
+async def test_options_write_posts_only_the_changed_token(coordinator) -> None:
+    href = '/course/vs/0'
+    coordinator._observe.apply(href, {
+        'x.com.samsung.da.options': ['DeviceType_0167', 'Course_16', 'GMT_04'],
+    }, source='poll')
+
+    desc = laundry.cycle_select(translation_key='dryer_cycle', icon='x')
+    bound = BoundEntity(href=href, capability=None, desc=desc)
+
+    await coordinator.async_send_command(bound, '1D')
+
+    posted_path, posted_bytes = coordinator._session.post_calls[0]
+    assert posted_path == ['course', 'vs', '0']
+    assert cbor2.loads(posted_bytes) == {'x.com.samsung.da.options': ['Course_1D']}
+
+
+async def test_options_write_optimistic_cache_keeps_sibling_tokens(coordinator) -> None:
+    """The regression this guards: applying the minimal wire body straight
+    to the cache would wipe DeviceType_0167/GMT_04 until the next poll."""
+    href = '/course/vs/0'
+    coordinator._observe.apply(href, {
+        'x.com.samsung.da.options': ['DeviceType_0167', 'Course_16', 'GMT_04'],
+    }, source='poll')
+
+    desc = laundry.cycle_select(translation_key='dryer_cycle', icon='x')
+    bound = BoundEntity(href=href, capability=None, desc=desc)
+
+    await coordinator.async_send_command(bound, '1D')
+
+    cached = coordinator._cache.get(href)
+    assert cached['x.com.samsung.da.options'] == [
+        'DeviceType_0167', 'Course_1D', 'GMT_04',
+    ]

tests/test_dishwasher_capabilities.py

@@ -28,9 +28,7 @@ class TestCycleOptions:
         rep = {'x.com.samsung.da.options': ['DeviceType_0001', 'Course_0E', 'GMT_04']}
         path, body = desc.write_fn('90', rep)
         assert path == ['course', 'vs', '0']
-        assert body == {
-            'x.com.samsung.da.options': ['DeviceType_0001', 'Course_90', 'GMT_04'],
-        }
+        assert body == {'x.com.samsung.da.options': ['Course_90']}
 
 
 class TestDishwasherOptions:

tests/test_fridge_capabilities.py

@@ -258,3 +258,39 @@ class TestRefrigeratorAiEnergyLevelFixtureCoverage:
         assert unbound == []
         state = flatten(bound, resources)
         assert state['ai_energy_level'] == '1'
+
+
+class TestSwitchOffIsNotInverted:
+    """Regression: several fridge SwitchDesc write_fns used ``'On' if p else 'Off'``.
+
+    The switch platform passes the literal string ``'Off'`` on turn-off (and
+    ``'On'`` on turn-on). ``'Off'`` is a non-empty, truthy string, so
+    ``'On' if p else 'Off'`` always evaluated to ``'On'`` — turning the switch
+    OFF silently re-sent ``On`` and the switch could never be turned off. The
+    guard must compare ``p == 'On'`` (as the other capability files already do).
+    """
+
+    # (switch descriptor, payload key it writes)
+    CASES = [
+        (fridge.ICEMAKER_NIGHTTIME.entities[0], 'ice.night.status'),
+        (fridge.STATUS_LOCK.entities[0], 'x.com.samsung.da.ado.devicecontrol'),
+        (fridge.STATUS_LOCK.entities[1], 'x.com.samsung.da.device.sound'),
+        (fridge.DEFROST_DELAY.entities[0], 'x.com.samsung.da.delayDefrost'),
+        (fridge.WELCOME_LIGHTING.entities[0], 'status'),
+        (fridge.CABINET_LIGHT.entities[1], 'light.dimming.status'),
+        (fridge.ICEMAKER_STATUS_FALLBACK.entities[0], 'x.com.samsung.da.iceMaker'),
+    ]
+
+    def test_turning_off_sends_off(self):
+        for desc, key in self.CASES:
+            _segs, payload = desc.write_fn('Off', {})
+            assert payload[key] == 'Off', (
+                f"{desc.key}: OFF must send 'Off', got {payload[key]!r} (inverted)"
+            )
+
+    def test_turning_on_sends_on(self):
+        for desc, key in self.CASES:
+            _segs, payload = desc.write_fn('On', {})
+            assert payload[key] == 'On', (
+                f"{desc.key}: ON must send 'On', got {payload[key]!r}"
+            )

tests/test_laundry_capabilities.py

@@ -153,14 +153,15 @@ class TestCycleSelect:
         live = {'/wm/editcourse/vs/0': {'x.com.samsung.da.editCourseList': 'EditCourseList_16'}}
         assert desc.exists_fn({}, live) is True
 
-    def test_cycle_write_rmw_on_options(self):
+    def test_cycle_write_is_single_token(self):
+        """Confirmed on real hardware (issue #54): the device merges a
+        single-token options[] write by prefix itself, so the write only
+        needs to carry the changed token, not the whole rewritten array."""
         desc = laundry.cycle_select(translation_key='dryer_cycle', icon='x')
         rep = {'x.com.samsung.da.options': ['DeviceType_0167', 'Course_16', 'GMT_04']}
         path, body = desc.write_fn('1D', rep)
         assert path == ['course', 'vs', '0']
-        assert body == {
-            'x.com.samsung.da.options': ['DeviceType_0167', 'Course_1D', 'GMT_04'],
-        }
+        assert body == {'x.com.samsung.da.options': ['Course_1D']}
 
     def test_cycle_write_noop_without_options(self):
         desc = laundry.cycle_select(translation_key='dryer_cycle', icon='x')

tests/test_oven_capabilities.py

@@ -137,7 +137,10 @@ def test_oven_mode_rejects_unknown():
 
 
 # ---------------------------------------------------------------------------
-# OVEN_MODE options-array RMW (lamp, sound, fast_preheat, natural_steam)
+# OVEN_MODE options-array writes (lamp, sound, fast_preheat, natural_steam).
+# Confirmed on real hardware (issue #54): a write only needs to carry the
+# changed token -- the device matches by prefix, evicts the stale token, and
+# merges the result into the array itself. No read-modify-write needed.
 # ---------------------------------------------------------------------------
 
 def _mode_rep(*extra_opts):
@@ -146,12 +149,11 @@ def _mode_rep(*extra_opts):
     ]}
 
 
-def test_lamp_write_replaces_slot():
+def test_lamp_write_is_single_token():
     desc = next(e for e in oven.OVEN_MODE.entities if e.key == 'lamp')
     path, body = desc.write_fn('On', _mode_rep())
-    opts = body['x.com.samsung.da.options']
-    assert 'UpperLamp_On' in opts
-    assert 'UpperLamp_Off' not in opts
+    assert path == ['mode', 'vs', '0']
+    assert body == {'x.com.samsung.da.options': ['UpperLamp_On']}
 
 
 def test_lamp_write_requires_existing_options():
@@ -159,20 +161,19 @@ def test_lamp_write_requires_existing_options():
     assert desc.write_fn('On', {}) is None
 
 
-def test_sound_write_preserves_other_options():
+def test_sound_write_is_single_token():
     desc = next(e for e in oven.OVEN_MODE.entities if e.key == 'sound')
     path, body = desc.write_fn('Off', _mode_rep())
-    opts = body['x.com.samsung.da.options']
-    assert 'Sound_Off' in opts
-    assert 'UpperLamp_Off' in opts     # other slot unchanged
+    assert body == {'x.com.samsung.da.options': ['Sound_Off']}
 
 
-def test_natural_steam_appended_if_absent():
-    """NaturalSteam slot is absent until first write — write_fn must append it."""
+def test_natural_steam_write_is_single_token():
+    """NaturalSteam's slot may be absent from the live rep until first
+    write -- the single-token write covers both the insert and replace case
+    identically, since the device merges by prefix either way."""
     desc = next(e for e in oven.OVEN_MODE.entities if e.key == 'natural_steam')
     path, body = desc.write_fn('On', _mode_rep())   # no NaturalSteam_* in rep
-    opts = body['x.com.samsung.da.options']
-    assert any(o.startswith('NaturalSteam_') for o in opts)
+    assert body == {'x.com.samsung.da.options': ['NaturalSteam_On']}
 
 
 # ---------------------------------------------------------------------------

tests/test_washer_capabilities.py

@@ -102,13 +102,14 @@ class TestWasherCourse:
         assert desc.exists_fn({}, live) is True
 
     def test_cycle_write(self):
+        """Confirmed on real hardware (issue #54): the write only needs to
+        carry the changed token -- the device matches by prefix, evicts the
+        stale token, and merges the result into the array itself."""
         desc = next(e for e in washer.WASHER_COURSE.entities if e.key == 'cycle')
         rep = {'x.com.samsung.da.options': ['DeviceType_0167', 'Course_1C', 'GMT_04']}
         path, body = desc.write_fn('1D', rep)
         assert path == ['course', 'vs', '0']
-        assert body == {
-            'x.com.samsung.da.options': ['DeviceType_0167', 'Course_1D', 'GMT_04'],
-        }
+        assert body == {'x.com.samsung.da.options': ['Course_1D']}
 
 
 class TestDrumClean:
@@ -230,20 +231,22 @@ class TestDetergentSoftenerDosing:
 
     def test_quantity_write(self):
         """The UI selects a padded supported code ('01'); the write posts the
-        un-padded device code ('1'), mirroring how the device reports it."""
+        un-padded device code ('1'), mirroring how the device reports it.
+
+        Confirmed on real hardware (issue #54): the write only needs to carry
+        the changed token -- the device matches by prefix, evicts the stale
+        token, and merges the result into the array itself. No need to read
+        the current array back and rewrite it whole."""
         rep = {'x.com.samsung.da.options': list(_DOSING_OPTIONS)}
         path, body = self._desc('detergent_quantity').write_fn('01', rep)
         assert path == ['course', 'vs', '0']
-        assert 'DetergentLevelCtrl_1' in body['x.com.samsung.da.options']
-        assert 'DetergentLevelCtrl_3' not in body['x.com.samsung.da.options']
-        # untouched siblings survive the read-modify-write
-        assert 'SoftenerLevelCtrl_3' in body['x.com.samsung.da.options']
+        assert body == {'x.com.samsung.da.options': ['DetergentLevelCtrl_1']}
 
     def test_hardness_write(self):
         rep = {'x.com.samsung.da.options': list(_DOSING_OPTIONS)}
         path, body = self._desc('softener_concentration').write_fn('03', rep)
         assert path == ['course', 'vs', '0']
-        assert 'SoftenerLevel2Ctrl_3' in body['x.com.samsung.da.options']
+        assert body == {'x.com.samsung.da.options': ['SoftenerLevel2Ctrl_3']}
 
     def test_low_reservoir_off_when_alarm_off(self):
         rep = {'x.com.samsung.da.options': _DOSING_OPTIONS}
@@ -305,18 +308,20 @@ class TestWashOptionToggles:
         """write_fn receives the same 'On'/'Off' string switch.py sends
         (not a bool) -- covers a bug where an earlier `'On' if p else 'Off'`
         implementation always wrote 'On', since any non-empty string
-        (including 'Off') is truthy."""
+        (including 'Off') is truthy.
+
+        The write body carries only the changed token (issue #54: confirmed
+        the device merges by prefix itself), not the whole options array."""
         for key in self._keys():
             prefix = self._prefix(key)
             rep = {'x.com.samsung.da.options': [f'{prefix}_Off', 'GMT_02']}
             path, body = self._desc(key).write_fn('On', rep)
             assert path == ['course', 'vs', '0']
-            assert f'{prefix}_On' in body['x.com.samsung.da.options']
-            assert 'GMT_02' in body['x.com.samsung.da.options']
+            assert body == {'x.com.samsung.da.options': [f'{prefix}_On']}
 
             rep = {'x.com.samsung.da.options': [f'{prefix}_On']}
             path, body = self._desc(key).write_fn('Off', rep)
-            assert f'{prefix}_Off' in body['x.com.samsung.da.options']
+            assert body == {'x.com.samsung.da.options': [f'{prefix}_Off']}
             assert f'{prefix}_On' not in body['x.com.samsung.da.options']
 
     def test_write_rejects_non_on_off_payload(self):