fix: don't let an in-progress settle window block a second write's own optimistic apply

Widening the settle window to tens of seconds (previous commit) made a
real gap much more likely to bite: apply() gated every source,
including 'optimistic', on _is_settling. /course/vs/0 backs several
independent washer selects (cycle, detergent quantity, softener
quantity, ...), so picking a second one while the first write's window
was still open -- routine within a ~43s window -- had its own
optimistic value silently dropped from the cache instead of shown,
recreating the exact "write doesn't seem to apply" symptom the guard
exists to prevent, just for whichever write lost the race.

Let source='optimistic' always bypass the gate; poll/sweep/observe
stay gated as before. mark_write_pending still re-arms the window
right after, so the newer write is protected going forward.

custom_components/localthings/observe.py

@@ -115,8 +115,22 @@ class ObserveManager:
         could each read the same prior rep and the second writer would
         silently lose the first's fields, reintroducing the exact bug
         this merge fixes.
+
+        `source == 'optimistic'` always bypasses the settle gate below,
+        even while this href is already settling from an earlier write.
+        Several independent selects can share one href -- issue #9's washer
+        packs cycle, detergent, and softener settings onto the same
+        /course/vs/0 -- so a second, different write to the same href
+        while the first's settle window is still open (up to
+        _POST_TIMEOUT_S + _POLL_TIMEOUT_S, tens of seconds -- see
+        coordinator.async_send_command) is an expected, normal sequence,
+        not a stale echo of the first write. Gating it the same as a
+        poll/sweep/observe update would silently drop the user's own
+        second selection from the cache until the first write's guard
+        happened to expire, i.e. the exact symptom this guard exists to
+        prevent, just relocated to whichever write loses the race.
         """
-        if self._is_settling(href):
+        if source != 'optimistic' and self._is_settling(href):
             self.log.debug("dropping %s update for %s (settling)", source, href)
             return False
         with self._cache_lock:

tests/localthings/test_coordinator.py

@@ -808,6 +808,45 @@ async def test_send_command_survives_stale_confirm_poll(
     assert coordinator._cache.get('/test/vs/0') == {'value': 5}
 
 
+async def test_second_write_to_same_href_lands_during_first_writes_settle_window(
+    hass: HomeAssistant, mock_entry, mock_coordinator_observe_session,
+) -> None:
+    """Regression for issue #9: /course/vs/0 backs several independent
+    washer selects (cycle, detergent quantity, softener quantity, ...)
+    sharing one href. The settle window is now sized to tens of seconds
+    (see test_send_command_survives_stale_confirm_poll), which makes a
+    second, different write to that href landing while the first write's
+    window is still open a routine occurrence -- e.g. a user picking a
+    cycle and then adjusting detergent quantity within the same minute --
+    not a rare edge case. The second write's own optimistic value must
+    still show up immediately; a guard meant to protect optimistic writes
+    from stale device echoes must not itself suppress a later one."""
+    from custom_components.localthings.registry.discovery import BoundEntity
+    from custom_components.localthings.registry.entities import NumberDesc
+
+    fake = mock_coordinator_observe_session
+    await hass.config_entries.async_setup(mock_entry.entry_id)
+    await hass.async_block_till_done()
+    coordinator: LocalThingsCoordinator = hass.data[DOMAIN][mock_entry.entry_id]
+
+    desc_a = NumberDesc(key='cycle', field='cycle',
+                        write_fn=lambda p, rep, href: (['test', 'vs', '0'], {'cycle': p}))
+    desc_b = NumberDesc(key='detergent', field='detergent',
+                        write_fn=lambda p, rep, href: (['test', 'vs', '0'], {'detergent': p}))
+    bound_a = BoundEntity(href='/test/vs/0', capability=coordinator.bound[0].capability, desc=desc_a)
+    bound_b = BoundEntity(href='/test/vs/0', capability=coordinator.bound[0].capability, desc=desc_b)
+
+    with patch.object(fake, 'subscribe'):
+        fake.post = lambda *a, **k: (0x44, b'')
+        await coordinator.async_send_command(bound_a, 'Eco')
+        assert coordinator._cache.get('/test/vs/0') == {'cycle': 'Eco'}
+
+        # Still well within the first write's settle window.
+        await coordinator.async_send_command(bound_b, 'High')
+
+    assert coordinator._cache.get('/test/vs/0') == {'cycle': 'Eco', 'detergent': 'High'}
+
+
 class TestRemoteControlEnabled:
     """remote_control_enabled (registry/capabilities/common.py) is the
     single source of truth for the /remotectrl on/off signal, shared by

tests/localthings/test_observe.py

@@ -67,6 +67,30 @@ def test_apply_drops_update_during_settle_window():
     assert mgr.cache.get('/oven/vs/0') == {'a': 1}
 
 
+def test_apply_optimistic_bypasses_an_in_progress_settle_window():
+    """Regression for issue #9: /course/vs/0 backs several independent
+    washer selects (cycle, detergent quantity, softener quantity, ...).
+    Picking a second one while the first's settle window is still open
+    (now sized to tens of seconds -- see coordinator._POST_TIMEOUT_S/
+    _POLL_TIMEOUT_S) is a normal sequence, not a stale echo of the first
+    write, and must land in the cache immediately -- not get silently
+    dropped by a guard that exists to protect optimistic writes, not
+    suppress them."""
+    mgr = _manager()
+    mgr.cache.apply_rep('/course/vs/0', {'Course': '1C', 'Detergent': '1'}, source='seed')
+    mgr.mark_write_pending('/course/vs/0', settle_s=30.0)
+
+    result = mgr.apply('/course/vs/0', {'Detergent': '2'}, source='optimistic')
+
+    assert result is True
+    assert mgr.cache.get('/course/vs/0') == {'Course': '1C', 'Detergent': '2'}
+
+    # A poll/sweep/observe update racing in right behind it is still
+    # gated -- the second write's own guard (re-armed by mark_write_pending,
+    # not exercised directly here) is what protects it going forward.
+    assert mgr.apply('/course/vs/0', {'Detergent': '1'}, source='poll') is False
+
+
 def test_apply_accepts_update_after_settle_window_elapses():
     mgr = _manager()
     mgr.mark_write_pending('/oven/vs/0', settle_s=0.05)