washer/dryer: hold progress/progress_percentage at Finish/100 for 5 min

Fixes #345. progress and progress_percentage were gated on machine_state
alone, falling straight to "Idle"/0 the instant state left 'active' --
but a washer can flip state away from 'active' within the same poll
interval progress reaches 'Finish' (more reliably than the same-family
dryer, per the report), so an automation watching for a real 'Finish'
value could go a whole cycle without ever observing one.

Implemented read-side, per-entity (sensor.py's new _apply_sticky),
generalizing the existing _hysteresis_value/_apply_hysteresis pattern
finish_time already uses, rather than writing a synthetic override back
into the coordinator's cache. That cache is this integration's record of
what the device actually said, and is read by several unrelated
consumers -- write_fn, validate_fn, diagnostics, the observe-mode sweep
comparison against /device/0, _completion_minutes' own stale-remainingTime
workaround -- all of which would otherwise see fabricated state.

A first pass gated the hold's arm condition on state=='active' AND
progress=='Finish' occurring in the same rep, mirroring _is_active. A
second review caught that this could make the whole fix a no-op on the
one device #345 reports it for: if state has already reset by the time
progress is ever observed at 'Finish' -- exactly what #345 describes --
the arm condition never fires. _just_finished now arms on progress==
'Finish' alone. That reopens the staleness risk the state check existed
to guard against (a progress field stuck at 'Finish' forever would then
arm forever too), so _apply_sticky is edge-triggered: only a fresh
False->True transition (re)starts the window, and expiry is still
checked on every call even while the condition keeps matching -- a
stuck value still won't hold past sticky_seconds.

Dropping the state requirement also exposed a second gap: the bypass
that lets a new cycle's own real progress override a stale hold was
keyed on machine_state=='active', so it missed a new cycle immediately
paused (e.g. adding a sock) -- machine_state isn't 'active' while
paused. It's keyed on a live, non-Finish progress code instead
(_live_progress_code), independent of state, same reasoning as
_just_finished. And since the bypass needs the *real* live value, not
whatever rep_fn's own (differently gated) result says, SensorDesc grew
sticky_live_fn alongside sticky_value_fn: rep_fn's progress gate still
shows "Idle" while paused, but the real progress value read ungated
must win over the hold regardless.

registry/entities.py: SensorDesc gains sticky_fn/sticky_value_fn/
sticky_live_fn/sticky_bypass_fn/sticky_seconds -- see sensor.py's
_apply_sticky docstring for the full contract.

registry/capabilities/operational.py: progress/progress_percentage's
rep_fn is unchanged; they gain the sticky_* wiring above. machine_state
and the Running binary sensor are untouched -- still gated on real-time
state (cycle_active now shares _is_active's rep_fn directly rather than
a duplicate inline copy), so they never claim the appliance is still
running once it isn't.

tests/test_operational_capability.py

@@ -1,6 +1,10 @@
 """Unit tests for operational state capabilities."""
 
-from custom_components.localthings.registry.capabilities.operational import OPERATIONAL_STATE
+from custom_components.localthings.registry.capabilities.operational import (
+    OPERATIONAL_STATE,
+    _just_finished,
+    _live_progress_code,
+)
 from custom_components.localthings.registry.entities import NumberDesc
 
 
@@ -11,6 +15,55 @@ def test_machine_state_maps_samsung_to_ocf():
     assert ms.value_fn("Ready") == "idle"
 
 
+class TestJustFinished:
+    """`_just_finished` is progress/progress_percentage's sticky_fn arm
+    condition (issue #345) -- see sensor.py's _apply_sticky."""
+
+    def test_true_when_progress_is_finish_while_active(self):
+        assert _just_finished(
+            {"x.com.samsung.da.state": "Run", "x.com.samsung.da.progress": "Finish"}
+        )
+
+    def test_true_even_once_state_has_already_left_active(self):
+        """The exact issue #345 scenario this exists for: a washer's
+        `state` can already read idle by the time `progress` is observed
+        at 'Finish' -- unlike _is_active, this must still arm."""
+        assert _just_finished(
+            {"x.com.samsung.da.state": "Ready", "x.com.samsung.da.progress": "Finish"}
+        )
+
+    def test_false_while_active_but_not_yet_finished(self):
+        assert not _just_finished(
+            {"x.com.samsung.da.state": "Run", "x.com.samsung.da.progress": "Spin"}
+        )
+
+    def test_false_when_progress_absent(self):
+        assert not _just_finished({"x.com.samsung.da.state": "Run"})
+
+
+class TestLiveProgressCode:
+    """`_live_progress_code` is progress/progress_percentage's
+    sticky_bypass_fn (issue #345) -- see sensor.py's _apply_sticky."""
+
+    def test_true_for_a_concrete_non_finish_code(self):
+        assert _live_progress_code({"x.com.samsung.da.progress": "Wash"})
+
+    def test_true_regardless_of_state(self):
+        """Not gated on machine_state -- a new cycle's own real progress
+        must win over a held hold even while paused (e.g. adding a sock
+        mid-hold), not just while actively running."""
+        assert _live_progress_code(
+            {"x.com.samsung.da.state": "Pause", "x.com.samsung.da.progress": "Wash"}
+        )
+
+    def test_false_for_finish(self):
+        assert not _live_progress_code({"x.com.samsung.da.progress": "Finish"})
+
+    def test_false_when_absent_or_none(self):
+        assert not _live_progress_code({})
+        assert not _live_progress_code({"x.com.samsung.da.progress": "None"})
+
+
 class TestProgressPercentage:
     """issue #9: device firmware leaves progressPercentage stale (e.g. '1')
     after a cycle ends instead of resetting it, so it must be gated on

tests/test_sensor_sticky.py

@@ -0,0 +1,272 @@
+"""Unit tests for LocalThingsSensor's sticky gate (issue #345).
+
+Modeled on test_sensor_hysteresis.py: a _FakeCoordinator with just enough
+surface for LocalThingsEntity/LocalThingsSensor, driven directly rather
+than through a full coordinator/HA setup. Unlike that file, `data` here
+is derived from the real `flatten()` over the same `resources` dict
+`resource()` serves -- a single source of truth, so a test can't hide a
+production desync bug by hand-computing the two independently.
+"""
+
+from __future__ import annotations
+
+import time
+from dataclasses import replace
+from typing import cast
+
+from custom_components.localthings.coordinator import LocalThingsCoordinator
+from custom_components.localthings.registry.adapter import flatten
+from custom_components.localthings.registry.capabilities.operational import OPERATIONAL_STATE
+from custom_components.localthings.registry.discovery import BoundEntity
+from custom_components.localthings.sensor import LocalThingsSensor
+
+_PROGRESS_DESC = next(e for e in OPERATIONAL_STATE.entities if e.key == "progress")
+_PROGRESS_PERCENTAGE_DESC = next(
+    e for e in OPERATIONAL_STATE.entities if e.key == "progress_percentage"
+)
+_MACHINE_STATE_DESC = next(e for e in OPERATIONAL_STATE.entities if e.key == "machine_state")
+
+_HREF = "/operational/state/vs/0"
+_ALL_BOUND = [
+    BoundEntity(href=_HREF, capability=OPERATIONAL_STATE, desc=desc)
+    for desc in OPERATIONAL_STATE.entities
+]
+
+
+class _FakeConfigEntry:
+    def __init__(self):
+        self.options: dict = {}
+
+
+class _FakeCoordinator:
+    """Just enough surface for LocalThingsEntity/LocalThingsSensor.
+
+    `resources` is the one source of truth (standing in for the real
+    coordinator's live cache); `data` and `resource()` are both derived
+    from it exactly as the real coordinator derives `.data` from
+    `flatten(self.bound, self._cache.snapshot())` and `.resource()` from
+    the same snapshot -- so a test can't drift them apart in a way
+    production couldn't.
+    """
+
+    def __init__(self):
+        self.device_serial = "TEST-SERIAL"
+        self.config_entry = _FakeConfigEntry()
+        self.resources: dict[str, dict] = {}
+
+    def resource(self, href: str) -> dict:
+        return self.resources.get(href) or {}
+
+    @property
+    def data(self) -> dict:
+        return flatten(_ALL_BOUND, self.resources)
+
+
+def _sensor(desc):
+    coordinator = _FakeCoordinator()
+    bound = BoundEntity(href=_HREF, capability=OPERATIONAL_STATE, desc=desc)
+    sensor = LocalThingsSensor(cast(LocalThingsCoordinator, coordinator), bound)
+    return sensor, coordinator
+
+
+def _apply(coordinator, **fields):
+    """Merge `fields` onto the href's existing rep, the same shallow
+    {**cached, **rep} merge ObserveManager.apply() does in production
+    (issue #27) -- so a field this call doesn't mention (e.g. a stale
+    `progress` the device didn't repeat) stays exactly as a real partial
+    update would leave it, rather than being silently wiped."""
+    href_fields = {f"x.com.samsung.da.{k}": v for k, v in fields.items()}
+    coordinator.resources[_HREF] = {**coordinator.resources.get(_HREF, {}), **href_fields}
+
+
+def _replace(coordinator, **fields):
+    """A full rep replacement -- for a device reporting fresh from
+    scratch (e.g. a full poll GET), unlike _apply's partial-update merge."""
+    coordinator.resources[_HREF] = {f"x.com.samsung.da.{k}": v for k, v in fields.items()}
+
+
+def test_holds_finish_after_state_leaves_active():
+    """The #345 regression this guards: progress used to fall straight to
+    'Idle' the instant machine_state left active, even right after
+    reporting Finish."""
+    sensor, coordinator = _sensor(_PROGRESS_DESC)
+
+    _replace(coordinator, state="Run", progress="Finish", progressPercentage="100")
+    assert sensor.native_value == "Finish"
+
+    _replace(coordinator, state="Ready")  # device has moved on
+    assert sensor.native_value == "Finish"
+
+
+def test_holds_finish_even_when_state_already_idle_at_first_observation():
+    """issue #345's actual reported failure mode: state can already read
+    idle by the time `progress: Finish` is ever observed (the report's
+    washer skips a Run+Finish moment the same-family dryer still shows) --
+    the hold must still arm from that single rep."""
+    sensor, coordinator = _sensor(_PROGRESS_DESC)
+
+    _replace(coordinator, state="Ready", progress="Finish", progressPercentage="100")
+    assert sensor.native_value == "Finish"
+
+    # Still held on a later poll, even once the device stops repeating it.
+    _replace(coordinator, state="Ready")
+    assert sensor.native_value == "Finish"
+
+
+def test_progress_percentage_holds_100_regardless_of_the_raw_field_at_finish():
+    """The hold pins 100 explicitly (sticky_value_fn), not whatever the
+    raw field happened to hold at the finish moment -- a device that
+    never populates progressPercentage at all, or (issue #9's failure
+    mode) leaves a stale non-100 value there, must still show 100 while
+    held, not None/unknown or the stale figure."""
+    sensor, coordinator = _sensor(_PROGRESS_PERCENTAGE_DESC)
+
+    _replace(coordinator, state="Run", progress="Finish")  # no progressPercentage at all
+    assert sensor.native_value == 100
+
+    _replace(coordinator, state="Ready")
+    assert sensor.native_value == 100
+
+
+def test_real_data_flows_through_unheld_while_active():
+    sensor, coordinator = _sensor(_PROGRESS_DESC)
+
+    _replace(coordinator, state="Run", progress="Spin")
+    assert sensor.native_value == "Spin"
+
+    _replace(coordinator, state="Run", progress="Rinse")
+    assert sensor.native_value == "Rinse"
+
+
+def test_never_finished_stays_idle():
+    """A device that never actually reached Finish (e.g. Stop pressed
+    mid-cycle) must not be held at some prior mid-cycle value."""
+    sensor, coordinator = _sensor(_PROGRESS_DESC)
+
+    _replace(coordinator, state="Run", progress="Spin")
+    assert sensor.native_value == "Spin"
+
+    _replace(coordinator, state="Ready")
+    assert sensor.native_value == "Idle"
+
+
+def test_a_new_cycle_starting_overrides_the_hold():
+    """Starting a second cycle must show its own real progress
+    immediately, not the held Finish from the previous one."""
+    sensor, coordinator = _sensor(_PROGRESS_DESC)
+
+    _replace(coordinator, state="Run", progress="Finish", progressPercentage="100")
+    assert sensor.native_value == "Finish"
+
+    _replace(coordinator, state="Ready")
+    assert sensor.native_value == "Finish"  # still held
+
+    _replace(coordinator, state="Run", progress="Wash")
+    assert sensor.native_value == "Wash"
+
+
+def test_a_paused_new_cycle_also_overrides_the_hold():
+    """Not just an actively-running new cycle: adding a sock and pausing
+    mid-cycle must also show the real, current progress rather than a
+    stale hold from the previous cycle -- machine_state isn't 'active'
+    while paused, so a bypass keyed on that alone would miss this."""
+    sensor, coordinator = _sensor(_PROGRESS_DESC)
+
+    _replace(coordinator, state="Run", progress="Finish", progressPercentage="100")
+    assert sensor.native_value == "Finish"
+    _replace(coordinator, state="Ready")
+    assert sensor.native_value == "Finish"  # still held
+
+    _replace(coordinator, state="Pause", progress="Wash")
+    assert sensor.native_value == "Wash"
+
+
+def test_hold_expires_after_sticky_seconds():
+    # A fresh desc (frozen dataclass -- replace(), not mutation, so the
+    # module-level _PROGRESS_DESC other tests share stays untouched) with
+    # a short real window -- consistent with this codebase's own
+    # settle-guard tests (test_observe.py's mark_write_pending(...,
+    # settle_s=0.05)) -- rather than a zero window, which exercises "held
+    # disabled" rather than "an armed hold actually expires".
+    desc = replace(_PROGRESS_DESC, sticky_seconds=0.05)
+    sensor, coordinator = _sensor(desc)
+
+    _replace(coordinator, state="Run", progress="Finish", progressPercentage="100")
+    assert sensor.native_value == "Finish"
+
+    _replace(coordinator, state="Ready")
+    assert sensor.native_value == "Finish"  # still within the window
+
+    time.sleep(0.1)
+    assert sensor.native_value == "Idle"
+
+
+def test_a_progress_stuck_at_finish_does_not_hold_open_the_window_forever():
+    """Edge-triggering: sticky_fn keeps matching on every poll if the
+    device's own `progress` field never resets on its own (the same class
+    of quirk _completion_minutes' remainingTime-freeze workaround already
+    documents) -- the hold must still expire on schedule from its first
+    sighting, not have its deadline pushed out by every subsequent poll
+    that still (correctly, from the device's perspective) reports Finish
+    while machine_state has already gone idle."""
+    desc = replace(_PROGRESS_DESC, sticky_seconds=0.05)
+    sensor, coordinator = _sensor(desc)
+
+    _replace(coordinator, state="Ready", progress="Finish")
+    assert sensor.native_value == "Finish"
+
+    time.sleep(0.03)
+    # Device still (incorrectly) reports Finish on every subsequent poll --
+    # must not restart the window.
+    _replace(coordinator, state="Ready", progress="Finish")
+    assert sensor.native_value == "Finish"
+
+    time.sleep(0.03)  # 0.06s total since the first sighting -- past 0.05s
+    _replace(coordinator, state="Ready", progress="Finish")
+    assert sensor.native_value == "Idle"
+
+
+def test_non_sticky_sensor_is_unaffected():
+    """machine_state has no sticky_fn -- reads straight through, unchanged."""
+    sensor, coordinator = _sensor(_MACHINE_STATE_DESC)
+    _replace(coordinator, state="Run")
+    assert sensor.native_value == "active"
+    _replace(coordinator, state="Ready")
+    assert sensor.native_value == "idle"
+
+
+def test_cycle_active_and_machine_state_are_never_held():
+    """The whole point of scoping the hold to progress/progress_percentage
+    only: cycle_active (Running) and machine_state must keep reflecting
+    the device's real-time state throughout, never claiming the appliance
+    is still running once it isn't."""
+    progress_sensor, coordinator = _sensor(_PROGRESS_DESC)
+    machine_state_bound = BoundEntity(
+        href=_HREF, capability=OPERATIONAL_STATE, desc=_MACHINE_STATE_DESC
+    )
+    machine_state_sensor = LocalThingsSensor(
+        cast(LocalThingsCoordinator, coordinator), machine_state_bound
+    )
+
+    _replace(coordinator, state="Run", progress="Finish", progressPercentage="100")
+    assert progress_sensor.native_value == "Finish"
+    assert machine_state_sensor.native_value == "active"
+
+    _replace(coordinator, state="Ready")
+    assert progress_sensor.native_value == "Finish"  # held
+    assert machine_state_sensor.native_value == "idle"  # real-time, unaffected
+
+
+def test_a_partial_update_that_omits_progress_does_not_erase_the_hold():
+    """A device's partial poll/notify that doesn't repeat `progress` at
+    all (issue #27's documented shape) merges onto the cache rather than
+    replacing it -- confirms the hold reads the merged rep, still showing
+    'Finish' from the earlier full rep, not a wiped/absent field."""
+    sensor, coordinator = _sensor(_PROGRESS_DESC)
+
+    _replace(coordinator, state="Run", progress="Finish", progressPercentage="100")
+    assert sensor.native_value == "Finish"
+
+    _apply(coordinator, state="Ready")  # partial merge, doesn't restate progress
+    assert coordinator.resources[_HREF]["x.com.samsung.da.progress"] == "Finish"
+    assert sensor.native_value == "Finish"