Make holding the session across a sequence the caller's choice

Holding _session_lock for a whole write sequence buys certainty about what
the appliance saw and when, but blocks every poll and entity write for the
sequence's full length -- up to 10 x 30s. Which of those matters more
depends on what is being probed, so it is now hold_session_lock on
async_raw_write_sequence and a field on the service, defaulting to the
holding behavior that shipped.

Off, the lock is taken per write and released across the settle waits, so
entities keep updating through a long sequence. Exactly one of the two
context managers is ever the real lock -- asyncio.Lock isn't reentrant.

Tests assert the lock's actual state during the settle wait in both modes,
rather than just that the flag is accepted.

README.md

@@ -123,7 +123,9 @@ data:
 
 Mind the shapes: what you write is sent verbatim, so the field names and types have to be the ones that resource actually uses. `/mode/vs/0` takes `modes` as an *array* on this board; a bare string, or the singular `mode`, is a different field the device will simply ignore. `read_resource` (below) with no `href` is the quickest way to see the real shape of everything before you write to any of it.
 
-Each write in `writes` (1-10 of them) needs `href` and a non-empty `payload`, sent verbatim as a partial-rep POST — this bypasses the remote-control-off block and every `write_fn`/`validate_fn` a normal entity write goes through, and sends exactly the fields you give it, so it can misconfigure your appliance if you get it wrong. `settle` (0-30s, default 0) is how long to wait *after* that write before starting the next one. The whole sequence runs under a single lock, so a routine poll can't land in the middle of it and blur which write is responsible for what the device does next.
+Each write in `writes` (1-10 of them) needs `href` and a non-empty `payload`, sent verbatim as a partial-rep POST — this bypasses the remote-control-off block and every `write_fn`/`validate_fn` a normal entity write goes through, and sends exactly the fields you give it, so it can misconfigure your appliance if you get it wrong. `settle` (0-30s, default 0) is how long to wait *after* that write before starting the next one.
+
+By default the whole sequence holds the device session from the first write to the last, settle delays included, so a routine poll or another entity's write can't land between two steps and blur which write the appliance was reacting to. The cost is that nothing else on that device updates until the sequence ends — up to 10 × 30s if you ask for the maximum of both. Set `hold_session_lock: false` to take the session per write and release it across the waits instead, trading that certainty for a device whose entities keep updating throughout.
 
 The response has one `results` entry per write, with `before`/`after` reps and a `changed` flag (every key/value in `payload` present and equal in the immediate readback):
 

custom_components/localthings/coordinator.py

@@ -1252,14 +1252,24 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
             )
 
     async def async_raw_write_sequence(
-        self, writes: list[dict], *, verify_after: float = 0.0
+        self,
+        writes: list[dict],
+        *,
+        verify_after: float = 0.0,
+        hold_session_lock: bool = True,
     ) -> dict[str, Any]:
         """Debug-only ordered multi-write (issue #300): a Samsung wall oven
         board discards settings writes while idle and only keeps them once
         a cycle is already running, which no single-write debug pass can
-        probe for. This owns the whole sequence -- every write shares one
-        `_session_lock` hold, so a poll can never interleave mid-sequence
-        and blur which write is responsible for what the device does next.
+        probe for.
+
+        `hold_session_lock` (default) keeps `_session_lock` for the whole
+        sequence, settle waits included, so nothing interleaves between
+        steps and blurs which write the appliance reacted to -- at the cost
+        of blocking polls and entity writes for the sequence's full length
+        (up to 10 x 30s). Pass False to take the lock per write and release
+        it across the waits, for a long sequence where a stalled poll costs
+        more than an interleaved read.
 
         `writes` are already on-the-wire hrefs: subdevice translation
         (canonical -> actual) is services.py's job, not this method's --
@@ -1288,13 +1298,20 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
 
         results: list[dict[str, Any]] = []
         last_payload_by_href: dict[str, dict] = {}
+        # Exactly one of these is the real lock, never both -- asyncio.Lock
+        # isn't reentrant, so nesting the same one would deadlock.
+        outer_lock = self._session_lock if hold_session_lock else contextlib.nullcontext()
         try:
-            async with self._session_lock:
+            async with outer_lock:
                 for i, (path_segs, href, payload, settle) in enumerate(parsed):
-                    before = self.resource(href)
-                    code, after = await self.hass.async_add_executor_job(
-                        self._raw_write_blocking, path_segs, payload, href
+                    per_write_lock = (
+                        contextlib.nullcontext() if hold_session_lock else self._session_lock
                     )
+                    async with per_write_lock:
+                        before = self.resource(href)
+                        code, after = await self.hass.async_add_executor_job(
+                            self._raw_write_blocking, path_segs, payload, href
+                        )
                     last_payload_by_href[href] = payload
                     results.append(
                         {
@@ -1307,10 +1324,9 @@ class LocalThingsCoordinator(DataUpdateCoordinator[dict[str, Any]]):
                             "changed": all(after.get(k) == v for k, v in payload.items()),
                         }
                     )
-                    # The lock is deliberately held across this wait, unlike
-                    # verify_after's below: a poll landing between two writes
-                    # blurs which one the appliance reacted to, which is what
-                    # the sequence exists to rule out. The caps bound it.
+                    # Under the default this wait happens inside the lock, so
+                    # nothing lands between two writes to blur which one the
+                    # appliance reacted to; see hold_session_lock above.
                     if settle and i < len(parsed) - 1:
                         await asyncio.sleep(settle)
         except Exception as err:

custom_components/localthings/services.py

@@ -31,6 +31,7 @@ ATTR_PAYLOAD = "payload"
 ATTR_SETTLE = "settle"
 ATTR_WRITES = "writes"
 ATTR_VERIFY_AFTER = "verify_after"
+ATTR_HOLD_SESSION_LOCK = "hold_session_lock"
 ATTR_DEVICE_ID = "device_id"
 
 _WRITE_ITEM_SCHEMA = vol.Schema(
@@ -58,6 +59,7 @@ _WRITE_RESOURCE_SCHEMA = vol.Schema(
         **cv.TARGET_SERVICE_FIELDS,
         vol.Required(ATTR_WRITES): vol.All(cv.ensure_list, [_WRITE_ITEM_SCHEMA]),
         vol.Optional(ATTR_VERIFY_AFTER): vol.Coerce(float),
+        vol.Optional(ATTR_HOLD_SESSION_LOCK): cv.boolean,
     }
 )
 
@@ -129,7 +131,9 @@ async def _async_write_resource(hass: HomeAssistant, call: ServiceCall) -> Servi
         for canonical, w in zip(canonicals, writes_in, strict=True)
     ]
     sequence = await coordinator.async_raw_write_sequence(
-        raw_writes, verify_after=call.data.get(ATTR_VERIFY_AFTER, 0.0)
+        raw_writes,
+        verify_after=call.data.get(ATTR_VERIFY_AFTER, 0.0),
+        hold_session_lock=call.data.get(ATTR_HOLD_SESSION_LOCK, True),
     )
 
     results = [

custom_components/localthings/services.yaml

@@ -32,6 +32,19 @@ write_resource:
         ["Bake"]}, "settle": 3}]
       selector:
         object:
+    hold_session_lock:
+      name: Hold the session for the whole sequence
+      description: >-
+        Keep the device session for the entire sequence, settle delays
+        included, so nothing else -- a routine poll, another entity's write
+        -- can land between two steps and blur which write the appliance
+        was reacting to. On by default. Turning it off takes the session
+        per write and frees it across the waits, which lets entities keep
+        updating during a long sequence at the cost of that certainty.
+      required: false
+      default: true
+      selector:
+        boolean:
     verify_after:
       name: Verify after
       description: >-

tests/test_services.py

@@ -188,6 +188,61 @@ async def test_write_resource_settle_honored_between_writes(hass, coordinator, d
     assert [call.args[0] for call in mock_sleep.call_args_list] == [2.0, 5.0]
 
 
+async def test_write_resource_holds_the_session_across_settle_by_default(
+    hass, coordinator, device_id
+):
+    """The default keeps the session for the whole sequence so nothing lands
+    between two writes -- asserted on the lock's real state during the wait,
+    not merely on the flag being accepted."""
+    coordinator._session = _FakeSession()
+    locked_during_settle = []
+
+    async def _record(_delay):
+        locked_during_settle.append(coordinator._session_lock.locked())
+
+    with patch(_SLEEP_TARGET, new=_record):
+        await _call_write(
+            hass,
+            device_id,
+            writes=[
+                {"href": "/a/vs/0", "payload": {"x": 1}, "settle": 2},
+                {"href": "/b/vs/0", "payload": {"x": 2}},
+            ],
+        )
+
+    assert locked_during_settle == [True]
+    # ...and it's handed back afterward, rather than leaked to the next call.
+    assert not coordinator._session_lock.locked()
+
+
+async def test_write_resource_releases_the_session_across_settle_when_asked(
+    hass, coordinator, device_id
+):
+    """hold_session_lock=False frees the session across the waits, so polls
+    and entity writes keep working through a long sequence."""
+    coordinator._session = _FakeSession()
+    locked_during_settle = []
+
+    async def _record(_delay):
+        locked_during_settle.append(coordinator._session_lock.locked())
+
+    with patch(_SLEEP_TARGET, new=_record):
+        response = await _call_write(
+            hass,
+            device_id,
+            writes=[
+                {"href": "/a/vs/0", "payload": {"x": 1}, "settle": 2},
+                {"href": "/b/vs/0", "payload": {"x": 2}},
+            ],
+            hold_session_lock=False,
+        )
+
+    assert locked_during_settle == [False]
+    assert not coordinator._session_lock.locked()
+    # Both writes still go out, in order -- only the locking differs.
+    assert [r["href"] for r in response["results"]] == ["/a/vs/0", "/b/vs/0"]
+
+
 async def test_write_resource_changed_true_when_readback_matches_payload(
     hass, coordinator, device_id
 ):