From f2af54d109f66cf0cb2679e88bea69fb63448628 Mon Sep 17 00:00:00 2001
From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com>
Date: Sat, 10 Oct 2026 11:00:37 +0200
Subject: [PATCH 1/4] feat(gateway_sessions): prioritize user commands over
 status polls (#680)

---
 custom_components/myhome/gateway_sessions.py | 181 ++++++++++-
 tests/test_gateway_sessions.py               | 307 +++++++++++++++++++
 2 files changed, 487 insertions(+), 1 deletion(-)

diff --git a/custom_components/myhome/gateway_sessions.py b/custom_components/myhome/gateway_sessions.py
index d07b112ba..13110da80 100644
--- a/custom_components/myhome/gateway_sessions.py
+++ b/custom_components/myhome/gateway_sessions.py
@@ -7,8 +7,11 @@
 from __future__ import annotations
 
 import asyncio
+import collections
 import contextlib
+import itertools
 import time
+from collections.abc import Iterator
 from typing import TYPE_CHECKING, Any
 
 from homeassistant.core import HomeAssistant
@@ -282,6 +285,182 @@ def close(self) -> None:
             self._event_watchdog = None
 
 
+class _PriorityQueueStorage:
+    """Multi-tier FIFO storage backing CommandPriorityQueue."""
+
+    def __init__(self) -> None:
+        """Initialize empty priority tiers."""
+        self._commands: collections.deque[Any] = collections.deque()
+        self._status: collections.deque[Any] = collections.deque()
+        self._others: collections.deque[Any] = collections.deque()
+
+    def __len__(self) -> int:
+        """Return total number of items across all tiers."""
+        return len(self._commands) + len(self._status) + len(self._others)
+
+    def __bool__(self) -> bool:
+        """Return True if any tier has items."""
+        return bool(self._commands or self._status or self._others)
+
+    def __iter__(self) -> Iterator[Any]:
+        """Iterate in priority order: commands, status, others."""
+        return itertools.chain(self._commands, self._status, self._others)
+
+    def __reversed__(self) -> Iterator[Any]:
+        """Iterate in reverse priority order."""
+        return itertools.chain(
+            reversed(self._others),
+            reversed(self._status),
+            reversed(self._commands),
+        )
+
+    def __repr__(self) -> str:
+        """Return string representation of priority tiers."""
+        return (
+            f"_PriorityQueueStorage(commands={list(self._commands)!r}, "
+            f"status={list(self._status)!r}, others={list(self._others)!r})"
+        )
+
+    def __contains__(self, item: Any) -> bool:
+        """Return True if item exists in any tier."""
+        return item in self._commands or item in self._status or item in self._others
+
+    def __getitem__(self, index: int | slice) -> Any:
+        """Get item by index or slice in priority order."""
+        if isinstance(index, slice):
+            return list(self)[index]
+        if not isinstance(index, int):
+            raise TypeError(f"queue indices must be integers or slices, not {type(index).__name__}")
+        total = len(self)
+        if index < 0:
+            index += total
+        if index < 0 or index >= total:
+            raise IndexError("queue index out of range")
+        if index < len(self._commands):
+            return self._commands[index]
+        index -= len(self._commands)
+        if index < len(self._status):
+            return self._status[index]
+        index -= len(self._status)
+        return self._others[index]
+
+    def append(self, item: Any) -> None:
+        """Append item to its corresponding priority tier."""
+        if item is None:
+            self._others.append(item)
+        elif isinstance(item, dict) and item.get("is_status_request", False):
+            self._status.append(item)
+        else:
+            self._commands.append(item)
+
+    def popleft(self) -> Any:
+        """Pop and return the highest priority item."""
+        if self._commands:
+            return self._commands.popleft()
+        if self._status:
+            return self._status.popleft()
+        if self._others:
+            return self._others.popleft()
+        raise IndexError("pop from an empty priority queue")
+
+    def pop(self) -> Any:
+        """Pop and return the lowest priority item from the tail."""
+        if self._others:
+            return self._others.pop()
+        if self._status:
+            return self._status.pop()
+        if self._commands:
+            return self._commands.pop()
+        raise IndexError("pop from an empty priority queue")
+
+    def remove(self, value: Any) -> None:
+        """Remove first occurrence of value."""
+        if value is None:
+            self._others.remove(value)
+            return
+        if isinstance(value, dict) and value.get("is_status_request", False):
+            try:
+                self._status.remove(value)
+                return
+            except ValueError:
+                pass
+        try:
+            self._commands.remove(value)
+        except ValueError:
+            try:
+                self._status.remove(value)
+            except ValueError:
+                self._others.remove(value)
+
+    def clear(self) -> None:
+        """Clear all priority tiers."""
+        self._commands.clear()
+        self._status.clear()
+        self._others.clear()
+
+    def copy(self) -> _PriorityQueueStorage:
+        """Return shallow copy of storage."""
+        new_storage = _PriorityQueueStorage()
+        new_storage._commands = self._commands.copy()
+        new_storage._status = self._status.copy()
+        new_storage._others = self._others.copy()
+        return new_storage
+
+    def count(self, value: Any) -> int:
+        """Return total occurrences of value across all tiers."""
+        return self._commands.count(value) + self._status.count(value) + self._others.count(value)
+
+
+class CommandPriorityQueue(asyncio.Queue[Any]):
+    """Asyncio queue prioritizing user commands over telemetry polls."""
+
+    _queue: _PriorityQueueStorage
+
+    def _init(self, maxsize: int) -> None:
+        """Initialize custom multi-tier priority queue storage."""
+        self._queue = _PriorityQueueStorage()
+
+    async def put(self, item: Any) -> None:
+        """Put an item into the queue, prioritizing command waiters when full."""
+        while self.full():
+            if getattr(self, "_is_shutdown", False):
+                raise getattr(asyncio, "QueueShutDown", Exception)
+            putter = self._get_loop().create_future()
+            is_status = isinstance(item, dict) and item.get("is_status_request", False)
+            is_cmd = not is_status and item is not None
+            setattr(putter, "_is_cmd", is_cmd)
+            if is_cmd:
+                idx = 0
+                for p in self._putters:
+                    if getattr(p, "_is_cmd", False):
+                        idx += 1
+                    else:
+                        break
+                self._putters.insert(idx, putter)
+            else:
+                self._putters.append(putter)
+            try:
+                await putter
+            except Exception:
+                putter.cancel()
+                with contextlib.suppress(ValueError):
+                    self._putters.remove(putter)
+                if not self.full() and not putter.cancelled():
+                    self._wakeup_next(self._putters)
+                raise
+        return self.put_nowait(item)
+
+    @property
+    def command_qsize(self) -> int:
+        """Return the number of queued user commands."""
+        return len(self._queue._commands)
+
+    @property
+    def status_qsize(self) -> int:
+        """Return the number of queued status requests."""
+        return len(self._queue._status)
+
+
 class CommandWorkerPool:
     """Manages the pool of command workers, send queue, and session pacing."""
 
@@ -301,7 +480,7 @@ def __init__(
             if hasattr(self.gateway, "profile") and self.gateway.profile
             else 250
         )
-        self.send_buffer: asyncio.Queue[Any] = asyncio.Queue(maxsize=queue_max_size)
+        self.send_buffer: CommandPriorityQueue = CommandPriorityQueue(maxsize=queue_max_size)
 
         self._event_session_ready = event_session_ready if event_session_ready is not None else asyncio.Event()
 
diff --git a/tests/test_gateway_sessions.py b/tests/test_gateway_sessions.py
index 733af9c44..7b89cbcb7 100644
--- a/tests/test_gateway_sessions.py
+++ b/tests/test_gateway_sessions.py
@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import asyncio
+from typing import Any
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -13,9 +14,11 @@
 from custom_components.myhome.gateway_sessions import (
     COMMAND_SESSION_IDLE_TIMEOUT,
     EVENT_STALL_TIMEOUT,
+    CommandPriorityQueue,
     CommandWorkerPool,
     EventSessionRunner,
     _cancel_written,
+    _PriorityQueueStorage,
     _resolve_written,
     _session_is_open,
 )
@@ -31,6 +34,7 @@ def mock_handler() -> MagicMock:
     handler.gateway = MagicMock(spec=OWNGateway)
     handler.gateway.profile = MagicMock()
     handler.gateway.profile.max_queue_size = 250
+    handler.gateway.profile.command_queue_delay = 0
     handler.command_session_idle_timeout = COMMAND_SESSION_IDLE_TIMEOUT
     handler.hass = MagicMock()
     handler._on_event_connection_state_change = MagicMock()
@@ -224,3 +228,306 @@ async def test_command_worker_pool_sending_loop_no_terminate_reset(mock_handler:
 
     # Must stay terminated
     assert pool._terminate_sender
+
+
+# ── CommandPriorityQueue & Storage ───────────────────────────────────────────
+
+
+async def test_command_priority_queue_ordering() -> None:
+    """User commands are dequeued strictly ahead of status polls, preserving FIFO within tiers."""
+    q: CommandPriorityQueue = CommandPriorityQueue()
+
+    s1 = {"message": "S1", "is_status_request": True}
+    s2 = {"message": "S2", "is_status_request": True}
+    c1 = {"message": "C1", "is_status_request": False}
+    s3 = {"message": "S3", "is_status_request": True}
+    c2 = {"message": "C2", "is_status_request": False}
+
+    for item in [s1, s2, c1, s3, c2]:
+        q.put_nowait(item)
+
+    assert q.qsize() == 5
+    assert q.command_qsize == 2
+    assert q.status_qsize == 3
+
+    # Inspection of internal sequence matches priority ordering:
+    assert [item["message"] for item in list(q._queue)] == ["C1", "C2", "S1", "S2", "S3"]
+
+    # Dequeue order strictly yields commands first (FIFO), then status polls (FIFO)
+    assert (await q.get())["message"] == "C1"
+    assert q.command_qsize == 1
+    assert q.status_qsize == 3
+
+    assert (await q.get())["message"] == "C2"
+    assert q.command_qsize == 0
+    assert q.status_qsize == 3
+
+    assert (await q.get())["message"] == "S1"
+    assert q.command_qsize == 0
+    assert q.status_qsize == 2
+
+    assert (await q.get())["message"] == "S2"
+    assert (await q.get())["message"] == "S3"
+    assert q.empty()
+    assert q.command_qsize == 0
+    assert q.status_qsize == 0
+
+
+async def test_command_priority_queue_read_back_after_write() -> None:
+    """Read-back status query queued after a command cannot overtake the command."""
+    q: CommandPriorityQueue = CommandPriorityQueue()
+
+    c1 = {"message": "*1*1*21##", "is_status_request": False}
+    s1 = {"message": "*#1*21##", "is_status_request": True}
+
+    await q.put(c1)
+    await q.put(s1)
+
+    first = await q.get()
+    second = await q.get()
+
+    assert first["message"] == "*1*1*21##"
+    assert first["is_status_request"] is False
+    assert second["message"] == "*#1*21##"
+    assert second["is_status_request"] is True
+
+
+async def test_command_priority_queue_sentinels() -> None:
+    """Worker shutdown sentinels (None) are dequeued only after draining all commands and status polls."""
+    q: CommandPriorityQueue = CommandPriorityQueue()
+
+    s1 = {"message": "S1", "is_status_request": True}
+    c1 = {"message": "C1", "is_status_request": False}
+    sentinel = None
+    c2 = {"message": "C2", "is_status_request": False}
+
+    await q.put(s1)
+    await q.put(c1)
+    await q.put(sentinel)
+    await q.put(c2)
+
+    assert [item if item is None else item["message"] for item in list(q._queue)] == [
+        "C1",
+        "C2",
+        "S1",
+        None,
+    ]
+
+    assert (await q.get())["message"] == "C1"
+    assert (await q.get())["message"] == "C2"
+    assert (await q.get())["message"] == "S1"
+    assert await q.get() is None
+    assert q.empty()
+
+
+async def test_command_priority_queue_full_backpressure() -> None:
+    """When full, waiting command putters take precedence over waiting status putters."""
+    q: CommandPriorityQueue = CommandPriorityQueue(maxsize=1)
+
+    s0 = {"message": "S0", "is_status_request": True}
+    s1 = {"message": "S1", "is_status_request": True}
+    c1 = {"message": "C1", "is_status_request": False}
+    c2 = {"message": "C2", "is_status_request": False}
+
+    # Queue full
+    await q.put(s0)
+    assert q.full()
+
+    admitted: list[str] = []
+
+    async def put_item(item: dict[str, Any]) -> None:
+        await q.put(item)
+        admitted.append(item["message"])
+
+    # Launch background puts: status first, then two commands
+    t_s1 = asyncio.create_task(put_item(s1))
+    await asyncio.sleep(0)
+
+    t_c1 = asyncio.create_task(put_item(c1))
+    await asyncio.sleep(0)
+
+    t_c2 = asyncio.create_task(put_item(c2))
+    await asyncio.sleep(0)
+
+    # Pop s0; c1 should be admitted next (ahead of s1)
+    item0 = await q.get()
+    assert item0["message"] == "S0"
+    await asyncio.sleep(0)
+    assert admitted == ["C1"]
+
+    # Pop c1; c2 should be admitted next (ahead of s1)
+    item1 = await q.get()
+    assert item1["message"] == "C1"
+    await asyncio.sleep(0)
+    assert admitted == ["C1", "C2"]
+
+    # Pop c2; s1 should be admitted next
+    item2 = await q.get()
+    assert item2["message"] == "C2"
+    await asyncio.sleep(0)
+    assert admitted == ["C1", "C2", "S1"]
+
+    # Pop s1; queue is now empty
+    item3 = await q.get()
+    assert item3["message"] == "S1"
+    assert q.empty()
+
+    await asyncio.gather(t_s1, t_c1, t_c2)
+
+
+async def test_command_priority_queue_backpressure_cancellation() -> None:
+    """Cancelled putters clean up cleanly and wake the next waiter in line."""
+    q: CommandPriorityQueue = CommandPriorityQueue(maxsize=1)
+
+    s0 = {"message": "S0", "is_status_request": True}
+    c1 = {"message": "C1", "is_status_request": False}
+    c2 = {"message": "C2", "is_status_request": False}
+
+    await q.put(s0)
+
+    t_c1 = asyncio.create_task(q.put(c1))
+    t_c2 = asyncio.create_task(q.put(c2))
+    await asyncio.sleep(0)
+
+    # Cancel c1 while waiting
+    t_c1.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await t_c1
+
+    # Pop s0; c2 should wake up
+    assert (await q.get())["message"] == "S0"
+    await asyncio.sleep(0)
+    assert t_c2.done()
+    assert (await q.get())["message"] == "C2"
+    assert q.empty()
+
+
+def test_command_priority_queue_storage_methods() -> None:
+    """Test sequence, mapping, and mutation methods of _PriorityQueueStorage."""
+    storage = _PriorityQueueStorage()
+    assert len(storage) == 0
+    assert not storage
+
+    c1 = {"message": "C1", "is_status_request": False}
+    c2 = {"message": "C2", "is_status_request": False}
+    s1 = {"message": "S1", "is_status_request": True}
+    sentinel = None
+
+    storage.append(s1)
+    storage.append(c1)
+    storage.append(sentinel)
+    storage.append(c2)
+
+    assert len(storage) == 4
+    assert bool(storage)
+    assert repr(storage).startswith("_PriorityQueueStorage(")
+
+    # Membership and count
+    assert c1 in storage
+    assert s1 in storage
+    assert sentinel in storage
+    assert "nonexistent" not in storage
+    assert storage.count(c1) == 1
+    assert storage.count("missing") == 0
+
+    # Indexing & slicing
+    assert storage[0] == c1
+    assert storage[1] == c2
+    assert storage[2] == s1
+    assert storage[3] is None
+    assert storage[-1] is None
+    assert storage[-4] == c1
+
+    with pytest.raises(IndexError):
+        _ = storage[4]
+    with pytest.raises(IndexError):
+        _ = storage[-5]
+    with pytest.raises(TypeError):
+        _ = storage["invalid"]  # type: ignore[index]
+
+    assert storage[1:3] == [c2, s1]
+    assert list(reversed(storage)) == [None, s1, c2, c1]
+
+    # Copy
+    copied = storage.copy()
+    assert len(copied) == 4
+    assert list(copied) == list(storage)
+
+    # Remove
+    storage.remove(c1)
+    assert len(storage) == 3
+    assert storage[0] == c2
+
+    storage.remove(s1)
+    assert len(storage) == 2
+
+    storage.remove(None)
+    assert len(storage) == 1
+    assert storage[0] == c2
+
+    with pytest.raises(ValueError):
+        storage.remove("not_present")
+
+    # Pop tail
+    assert storage.pop() == c2
+    assert len(storage) == 0
+
+    # Pop on empty raises IndexError
+    with pytest.raises(IndexError):
+        storage.popleft()
+    with pytest.raises(IndexError):
+        storage.pop()
+
+    # Clear
+    copied.clear()
+    assert len(copied) == 0
+
+
+async def test_worker_sends_user_command_ahead_of_pending_status_polls(
+    mock_handler: MagicMock,
+) -> None:
+    """Worker sends high-priority user commands before preceding queued status polls."""
+    pool = CommandWorkerPool(mock_handler)
+    pool._event_session_ready.set()
+
+    mock_session = MagicMock()
+    mock_session.connect = AsyncMock(return_value={"Success": True})
+    mock_session.is_connected = True
+    mock_session.close = AsyncMock()
+    mock_session._stream_reader = object()
+    mock_session._stream_writer = object()
+
+    sent_frames: list[str] = []
+
+    async def mock_send(message: Any, is_status_request: bool = False) -> list[Any]:
+        sent_frames.append(str(message))
+        return []
+
+    mock_session.send = AsyncMock(side_effect=mock_send)
+
+    with patch(
+        "custom_components.myhome.gateway.OWNCommandSession",
+        return_value=mock_session,
+    ):
+        # Queue 5 status requests
+        for i in range(1, 6):
+            await pool.send_status_request(OWNCommand(f"*#1*{i}##"))
+
+        # Queue 1 user command
+        await pool.send(OWNCommand("*1*1*21##"))
+
+        # Put sentinel to stop worker after draining
+        await pool.send_buffer.put(None)
+
+        # Run sending loop
+        await pool.sending_loop(0)
+
+    # Assert user command was sent FIRST before any of the status queries
+    assert sent_frames[0] == "*1*1*21##"
+    assert sent_frames[1:] == [
+        "*#1*1##",
+        "*#1*2##",
+        "*#1*3##",
+        "*#1*4##",
+        "*#1*5##",
+    ]

From 44e04b162605a7cad18dd2b6256a667ad74c630b Mon Sep 17 00:00:00 2001
From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com>
Date: Sat, 10 Oct 2026 11:13:18 +0200
Subject: [PATCH 2/4] fix(gateway_sessions): handle putter cancellation
 cleanly, add deque methods, and refine tier classification (#680)

---
 custom_components/myhome/gateway_sessions.py | 109 +++++++++++++++----
 tests/test_gateway_sessions.py               |  86 ++++++++++++---
 2 files changed, 159 insertions(+), 36 deletions(-)

diff --git a/custom_components/myhome/gateway_sessions.py b/custom_components/myhome/gateway_sessions.py
index 13110da80..c127b9d63 100644
--- a/custom_components/myhome/gateway_sessions.py
+++ b/custom_components/myhome/gateway_sessions.py
@@ -11,7 +11,7 @@
 import contextlib
 import itertools
 import time
-from collections.abc import Iterator
+from collections.abc import Iterable, Iterator
 from typing import TYPE_CHECKING, Any
 
 from homeassistant.core import HomeAssistant
@@ -285,6 +285,21 @@ def close(self) -> None:
             self._event_watchdog = None
 
 
+_TIER_COMMANDS = 0
+_TIER_STATUS = 1
+_TIER_OTHERS = 2
+
+
+def _classify_tier(item: Any) -> int:
+    """Classify item into priority tier: 0 (commands), 1 (status), 2 (others/sentinels)."""
+    if isinstance(item, dict):
+        if item.get("is_status_request", False):
+            return _TIER_STATUS
+        if "message" in item or "is_status_request" in item:
+            return _TIER_COMMANDS
+    return _TIER_OTHERS
+
+
 class _PriorityQueueStorage:
     """Multi-tier FIFO storage backing CommandPriorityQueue."""
 
@@ -321,6 +336,18 @@ def __repr__(self) -> str:
             f"status={list(self._status)!r}, others={list(self._others)!r})"
         )
 
+    def __eq__(self, other: object) -> bool:
+        """Return True if other has identical items in the same priority order."""
+        if isinstance(other, _PriorityQueueStorage):
+            return (
+                self._commands == other._commands
+                and self._status == other._status
+                and self._others == other._others
+            )
+        if isinstance(other, (collections.deque, list)):
+            return list(self) == list(other)
+        return False
+
     def __contains__(self, item: Any) -> bool:
         """Return True if item exists in any tier."""
         return item in self._commands or item in self._status or item in self._others
@@ -346,12 +373,33 @@ def __getitem__(self, index: int | slice) -> Any:
 
     def append(self, item: Any) -> None:
         """Append item to its corresponding priority tier."""
-        if item is None:
-            self._others.append(item)
-        elif isinstance(item, dict) and item.get("is_status_request", False):
+        tier = _classify_tier(item)
+        if tier == _TIER_COMMANDS:
+            self._commands.append(item)
+        elif tier == _TIER_STATUS:
             self._status.append(item)
         else:
-            self._commands.append(item)
+            self._others.append(item)
+
+    def appendleft(self, item: Any) -> None:
+        """Append item to the head of its corresponding priority tier."""
+        tier = _classify_tier(item)
+        if tier == _TIER_COMMANDS:
+            self._commands.appendleft(item)
+        elif tier == _TIER_STATUS:
+            self._status.appendleft(item)
+        else:
+            self._others.appendleft(item)
+
+    def extend(self, iterable: Iterable[Any]) -> None:
+        """Extend storage by appending elements from the iterable."""
+        for item in iterable:
+            self.append(item)
+
+    def extendleft(self, iterable: Iterable[Any]) -> None:
+        """Extend storage by prepending elements from the iterable."""
+        for item in iterable:
+            self.appendleft(item)
 
     def popleft(self) -> Any:
         """Pop and return the highest priority item."""
@@ -375,22 +423,28 @@ def pop(self) -> Any:
 
     def remove(self, value: Any) -> None:
         """Remove first occurrence of value."""
-        if value is None:
-            self._others.remove(value)
-            return
-        if isinstance(value, dict) and value.get("is_status_request", False):
-            try:
-                self._status.remove(value)
-                return
-            except ValueError:
-                pass
+        tier = _classify_tier(value)
+        if tier == _TIER_COMMANDS:
+            target = self._commands
+        elif tier == _TIER_STATUS:
+            target = self._status
+        else:
+            target = self._others
+
         try:
-            self._commands.remove(value)
+            target.remove(value)
+            return
         except ValueError:
-            try:
-                self._status.remove(value)
-            except ValueError:
-                self._others.remove(value)
+            pass
+
+        for q in (self._commands, self._status, self._others):
+            if q is not target:
+                try:
+                    q.remove(value)
+                    return
+                except ValueError:
+                    pass
+        raise ValueError(f"{value!r} not in priority queue")
 
     def clear(self) -> None:
         """Clear all priority tiers."""
@@ -410,6 +464,13 @@ def count(self, value: Any) -> int:
         """Return total occurrences of value across all tiers."""
         return self._commands.count(value) + self._status.count(value) + self._others.count(value)
 
+    def index(self, value: Any, start: int = 0, stop: int | None = None) -> int:
+        """Return first index of value."""
+        items = list(self)
+        if stop is None:
+            return items.index(value, start)
+        return items.index(value, start, stop)
+
 
 class CommandPriorityQueue(asyncio.Queue[Any]):
     """Asyncio queue prioritizing user commands over telemetry polls."""
@@ -426,8 +487,7 @@ async def put(self, item: Any) -> None:
             if getattr(self, "_is_shutdown", False):
                 raise getattr(asyncio, "QueueShutDown", Exception)
             putter = self._get_loop().create_future()
-            is_status = isinstance(item, dict) and item.get("is_status_request", False)
-            is_cmd = not is_status and item is not None
+            is_cmd = _classify_tier(item) == _TIER_COMMANDS
             setattr(putter, "_is_cmd", is_cmd)
             if is_cmd:
                 idx = 0
@@ -441,7 +501,7 @@ async def put(self, item: Any) -> None:
                 self._putters.append(putter)
             try:
                 await putter
-            except Exception:
+            except BaseException:
                 putter.cancel()
                 with contextlib.suppress(ValueError):
                     self._putters.remove(putter)
@@ -460,6 +520,11 @@ def status_qsize(self) -> int:
         """Return the number of queued status requests."""
         return len(self._queue._status)
 
+    @property
+    def other_qsize(self) -> int:
+        """Return the number of queued sentinels and unclassified items."""
+        return len(self._queue._others)
+
 
 class CommandWorkerPool:
     """Manages the pool of command workers, send queue, and session pacing."""
diff --git a/tests/test_gateway_sessions.py b/tests/test_gateway_sessions.py
index 7b89cbcb7..6ed1b34ab 100644
--- a/tests/test_gateway_sessions.py
+++ b/tests/test_gateway_sessions.py
@@ -293,30 +293,39 @@ async def test_command_priority_queue_read_back_after_write() -> None:
 
 
 async def test_command_priority_queue_sentinels() -> None:
-    """Worker shutdown sentinels (None) are dequeued only after draining all commands and status polls."""
+    """Worker shutdown sentinels (None and custom objects) are dequeued only after draining all commands and polls."""
     q: CommandPriorityQueue = CommandPriorityQueue()
 
     s1 = {"message": "S1", "is_status_request": True}
     c1 = {"message": "C1", "is_status_request": False}
     sentinel = None
+    custom_sentinel = "WORKER_TERMINATE"
     c2 = {"message": "C2", "is_status_request": False}
 
     await q.put(s1)
     await q.put(c1)
     await q.put(sentinel)
+    await q.put(custom_sentinel)
     await q.put(c2)
 
-    assert [item if item is None else item["message"] for item in list(q._queue)] == [
+    assert q.command_qsize == 2
+    assert q.status_qsize == 1
+    assert q.other_qsize == 2
+    assert q.qsize() == 5
+
+    assert [item if item in (None, custom_sentinel) else item["message"] for item in list(q._queue)] == [
         "C1",
         "C2",
         "S1",
         None,
+        custom_sentinel,
     ]
 
     assert (await q.get())["message"] == "C1"
     assert (await q.get())["message"] == "C2"
     assert (await q.get())["message"] == "S1"
     assert await q.get() is None
+    assert await q.get() == "WORKER_TERMINATE"
     assert q.empty()
 
 
@@ -389,17 +398,48 @@ async def test_command_priority_queue_backpressure_cancellation() -> None:
     t_c2 = asyncio.create_task(q.put(c2))
     await asyncio.sleep(0)
 
-    # Cancel c1 while waiting
+    assert len(q._putters) == 2
+
+    # Cancel c1 while waiting: it must be cleanly removed from _putters
     t_c1.cancel()
     with pytest.raises(asyncio.CancelledError):
         await t_c1
 
+    assert len(q._putters) == 1
+
     # Pop s0; c2 should wake up
     assert (await q.get())["message"] == "S0"
     await asyncio.sleep(0)
     assert t_c2.done()
     assert (await q.get())["message"] == "C2"
     assert q.empty()
+    assert len(q._putters) == 0
+
+
+async def test_command_priority_queue_woken_putter_cancellation_unblocks_next() -> None:
+    """A putter cancelled after being woken up by get() does not lose the slot for subsequent waiters."""
+    q: CommandPriorityQueue = CommandPriorityQueue(maxsize=1)
+
+    await q.put({"message": "S0", "is_status_request": True})
+
+    t1 = asyncio.create_task(q.put({"message": "C1", "is_status_request": False}))
+    t2 = asyncio.create_task(q.put({"message": "C2", "is_status_request": False}))
+    await asyncio.sleep(0)
+
+    # Consumer pops S0, which unblocks t1
+    val = q.get_nowait()
+    assert val["message"] == "S0"
+
+    # t1 is cancelled immediately upon waking
+    t1.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await t1
+
+    # t2 must NOT deadlock; it must be unblocked and successfully admit C2
+    done, _ = await asyncio.wait([t2], timeout=0.5)
+    assert t2 in done
+    assert (await q.get())["message"] == "C2"
+    assert q.empty()
 
 
 def test_command_priority_queue_storage_methods() -> None:
@@ -412,13 +452,15 @@ def test_command_priority_queue_storage_methods() -> None:
     c2 = {"message": "C2", "is_status_request": False}
     s1 = {"message": "S1", "is_status_request": True}
     sentinel = None
+    custom = "custom_obj"
 
     storage.append(s1)
     storage.append(c1)
     storage.append(sentinel)
     storage.append(c2)
+    storage.append(custom)
 
-    assert len(storage) == 4
+    assert len(storage) == 5
     assert bool(storage)
     assert repr(storage).startswith("_PriorityQueueStorage(")
 
@@ -426,6 +468,7 @@ def test_command_priority_queue_storage_methods() -> None:
     assert c1 in storage
     assert s1 in storage
     assert sentinel in storage
+    assert custom in storage
     assert "nonexistent" not in storage
     assert storage.count(c1) == 1
     assert storage.count("missing") == 0
@@ -435,36 +478,51 @@ def test_command_priority_queue_storage_methods() -> None:
     assert storage[1] == c2
     assert storage[2] == s1
     assert storage[3] is None
-    assert storage[-1] is None
-    assert storage[-4] == c1
+    assert storage[4] == custom
+    assert storage[-1] == custom
+    assert storage[-5] == c1
 
     with pytest.raises(IndexError):
-        _ = storage[4]
+        _ = storage[5]
     with pytest.raises(IndexError):
-        _ = storage[-5]
+        _ = storage[-6]
     with pytest.raises(TypeError):
         _ = storage["invalid"]  # type: ignore[index]
 
     assert storage[1:3] == [c2, s1]
-    assert list(reversed(storage)) == [None, s1, c2, c1]
+    assert list(reversed(storage)) == [custom, None, s1, c2, c1]
+    assert storage.index(c2) == 1
+    assert storage.index(custom) == 4
 
-    # Copy
+    # Copy and equality
     copied = storage.copy()
-    assert len(copied) == 4
+    assert len(copied) == 5
     assert list(copied) == list(storage)
+    assert copied == storage
+
+    # Extend and extendleft
+    c3 = {"message": "C3", "is_status_request": False}
+    c0 = {"message": "C0", "is_status_request": False}
+    copied.extend([c3])
+    assert copied[2] == c3
+    copied.appendleft(c0)
+    assert copied[0] == c0
 
     # Remove
     storage.remove(c1)
-    assert len(storage) == 3
+    assert len(storage) == 4
     assert storage[0] == c2
 
     storage.remove(s1)
-    assert len(storage) == 2
+    assert len(storage) == 3
 
     storage.remove(None)
-    assert len(storage) == 1
+    assert len(storage) == 2
     assert storage[0] == c2
 
+    storage.remove(custom)
+    assert len(storage) == 1
+
     with pytest.raises(ValueError):
         storage.remove("not_present")
 

From 8d1f7069f651172e93af16d9d0945caee60d47d9 Mon Sep 17 00:00:00 2001
From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com>
Date: Sat, 10 Oct 2026 11:45:13 +0200
Subject: [PATCH 3/4] fix(gateway_sessions): satisfy strict-typing ratchet and
 achieve 100% test coverage (#680)

---
 custom_components/myhome/gateway_sessions.py |  46 +++---
 tests/test_gateway_sessions.py               | 152 +++++++++++++++++--
 2 files changed, 167 insertions(+), 31 deletions(-)

diff --git a/custom_components/myhome/gateway_sessions.py b/custom_components/myhome/gateway_sessions.py
index c127b9d63..97562e3d7 100644
--- a/custom_components/myhome/gateway_sessions.py
+++ b/custom_components/myhome/gateway_sessions.py
@@ -424,26 +424,19 @@ def pop(self) -> Any:
     def remove(self, value: Any) -> None:
         """Remove first occurrence of value."""
         tier = _classify_tier(value)
-        if tier == _TIER_COMMANDS:
-            target = self._commands
-        elif tier == _TIER_STATUS:
-            target = self._status
-        else:
-            target = self._others
-
-        try:
-            target.remove(value)
-            return
-        except ValueError:
-            pass
-
-        for q in (self._commands, self._status, self._others):
-            if q is not target:
-                try:
-                    q.remove(value)
-                    return
-                except ValueError:
-                    pass
+        tiers = (
+            (self._commands, self._status, self._others)
+            if tier == _TIER_COMMANDS
+            else (self._status, self._commands, self._others)
+            if tier == _TIER_STATUS
+            else (self._others, self._commands, self._status)
+        )
+        for q in tiers:
+            try:
+                q.remove(value)
+                return
+            except ValueError:
+                pass
         raise ValueError(f"{value!r} not in priority queue")
 
     def clear(self) -> None:
@@ -476,17 +469,26 @@ class CommandPriorityQueue(asyncio.Queue[Any]):
     """Asyncio queue prioritizing user commands over telemetry polls."""
 
     _queue: _PriorityQueueStorage
+    _putters: collections.deque[asyncio.Future[Any]]
 
     def _init(self, maxsize: int) -> None:
         """Initialize custom multi-tier priority queue storage."""
         self._queue = _PriorityQueueStorage()
 
+    def _wakeup_next(self, waiters: collections.deque[asyncio.Future[Any]]) -> None:
+        """Wake up the next waiter that is not cancelled."""
+        while waiters:
+            waiter = waiters.popleft()
+            if not waiter.done():
+                waiter.set_result(None)
+                break
+
     async def put(self, item: Any) -> None:
         """Put an item into the queue, prioritizing command waiters when full."""
         while self.full():
             if getattr(self, "_is_shutdown", False):
                 raise getattr(asyncio, "QueueShutDown", Exception)
-            putter = self._get_loop().create_future()
+            putter: asyncio.Future[None] = asyncio.get_running_loop().create_future()
             is_cmd = _classify_tier(item) == _TIER_COMMANDS
             setattr(putter, "_is_cmd", is_cmd)
             if is_cmd:
@@ -545,7 +547,7 @@ def __init__(
             if hasattr(self.gateway, "profile") and self.gateway.profile
             else 250
         )
-        self.send_buffer: CommandPriorityQueue = CommandPriorityQueue(maxsize=queue_max_size)
+        self.send_buffer: asyncio.Queue[Any] = CommandPriorityQueue(maxsize=queue_max_size)
 
         self._event_session_ready = event_session_ready if event_session_ready is not None else asyncio.Event()
 
diff --git a/tests/test_gateway_sessions.py b/tests/test_gateway_sessions.py
index 6ed1b34ab..442a0ae77 100644
--- a/tests/test_gateway_sessions.py
+++ b/tests/test_gateway_sessions.py
@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import asyncio
+import collections
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -174,6 +175,90 @@ async def test_event_session_runner_read_session_auth_refused(mock_handler: Magi
     mock_session.get_next.assert_not_called()
 
 
+async def test_event_session_runner_retry_loop_and_backoff(mock_handler: MagicMock) -> None:
+    """_listening_loop retries _run_event_session with backoff when it returns True."""
+    runner = EventSessionRunner(mock_handler)
+    calls = 0
+
+    async def mock_run() -> bool:
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            return True
+        return False
+
+    runner._run_event_session = AsyncMock(side_effect=mock_run)
+
+    with patch("asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
+        await runner.listening_loop()
+        assert calls == 2
+        mock_sleep.assert_called_once()
+
+
+async def test_event_session_runner_run_event_session_terminated_or_exceptions(
+    mock_handler: MagicMock,
+) -> None:
+    """_run_event_session handles termination, TimeoutError stall, and generic exceptions."""
+    runner = EventSessionRunner(mock_handler)
+
+    # Terminated early return
+    runner._terminate_listener = True
+    assert await runner._run_event_session() is False
+    runner._terminate_listener = False
+
+    # Watchdog TimeoutError stall
+    with patch.object(runner, "_read_event_session", side_effect=TimeoutError("stall")):
+        with patch("asyncio.timeout") as mock_timeout:
+            mock_ctx = MagicMock()
+            mock_ctx.expired.return_value = True
+            mock_ctx.__aenter__ = AsyncMock(return_value=mock_ctx)
+            mock_ctx.__aexit__ = AsyncMock(side_effect=TimeoutError("stall"))
+            mock_timeout.return_value = mock_ctx
+            assert await runner._run_event_session() is True
+
+    # Generic exception
+    with patch.object(runner, "_read_event_session", side_effect=RuntimeError("boom")):
+        assert await runner._run_event_session() is True
+
+
+async def test_event_session_runner_read_session_reachable_change_and_process_error(
+    mock_handler: MagicMock,
+) -> None:
+    """_read_event_session logs reachable transitions and survives _process_message exceptions."""
+    runner = EventSessionRunner(mock_handler)
+    mock_session = MagicMock()
+    mock_session.connect = AsyncMock(return_value={"Success": True})
+    mock_session.is_connected = True
+    mock_session._stream_reader = object()
+    mock_session._stream_writer = object()
+
+    msg = OWNMessage("*1*1*11##")
+    mock_handler._process_message.side_effect = ValueError("bad handler")
+
+    step = 0
+
+    async def get_next_side_effect() -> OWNMessage | None:
+        nonlocal step
+        step += 1
+        if step == 1:
+            return msg
+        if step == 2:
+            mock_session.is_connected = False
+            mock_session._stream_reader = None
+            return None
+        if step == 3:
+            mock_session.is_connected = True
+            mock_session._stream_reader = object()
+            return None
+        runner._terminate_listener = True
+        return None
+
+    mock_session.get_next = AsyncMock(side_effect=get_next_side_effect)
+
+    await runner._read_event_session(mock_session)
+    assert step == 4
+
+
 # ── CommandWorkerPool ──────────────────────────────────────────────────────────
 
 
@@ -492,6 +577,7 @@ def test_command_priority_queue_storage_methods() -> None:
     assert storage[1:3] == [c2, s1]
     assert list(reversed(storage)) == [custom, None, s1, c2, c1]
     assert storage.index(c2) == 1
+    assert storage.index(c2, 0, 3) == 1
     assert storage.index(custom) == 4
 
     # Copy and equality
@@ -499,14 +585,31 @@ def test_command_priority_queue_storage_methods() -> None:
     assert len(copied) == 5
     assert list(copied) == list(storage)
     assert copied == storage
+    assert storage == list(storage)
+    assert storage == collections.deque(list(storage))
+    assert storage != "not_a_queue"
+    assert storage != [c1]
 
-    # Extend and extendleft
-    c3 = {"message": "C3", "is_status_request": False}
+    # Appendleft across all tiers
     c0 = {"message": "C0", "is_status_request": False}
-    copied.extend([c3])
-    assert copied[2] == c3
+    s0 = {"message": "S0", "is_status_request": True}
+    o0 = "other_0"
     copied.appendleft(c0)
     assert copied[0] == c0
+    copied.appendleft(s0)
+    assert s0 in copied
+    copied.appendleft(o0)
+    assert o0 in copied
+
+    # Extend and extendleft
+    c3 = {"message": "C3", "is_status_request": False}
+    s3 = {"message": "S3", "is_status_request": True}
+    o3 = "other_3"
+    copied.extend([c3])
+    assert c3 in copied
+    copied.extendleft([c3, s3, o3])
+    assert s3 in copied
+    assert o3 in copied
 
     # Remove
     storage.remove(c1)
@@ -526,21 +629,52 @@ def test_command_priority_queue_storage_methods() -> None:
     with pytest.raises(ValueError):
         storage.remove("not_present")
 
-    # Pop tail
-    assert storage.pop() == c2
-    assert len(storage) == 0
+    # Pop tail across all tiers: _others, _status, and _commands
+    pop_storage = _PriorityQueueStorage()
+    pop_storage.append({"message": "CMD", "is_status_request": False})
+    pop_storage.append({"message": "STAT", "is_status_request": True})
+    pop_storage.append("OTHER")
+
+    assert pop_storage.pop() == "OTHER"  # pops from _others
+    assert pop_storage.pop()["message"] == "STAT"  # pops from _status
+    assert pop_storage.pop()["message"] == "CMD"  # pops from _commands
 
     # Pop on empty raises IndexError
     with pytest.raises(IndexError):
-        storage.popleft()
+        pop_storage.popleft()
     with pytest.raises(IndexError):
-        storage.pop()
+        pop_storage.pop()
 
     # Clear
     copied.clear()
     assert len(copied) == 0
 
 
+async def test_command_priority_queue_shutdown_and_wakeup_skip() -> None:
+    """Queue shutdown raises exception and _wakeup_next skips completed waiters."""
+    q: CommandPriorityQueue = CommandPriorityQueue(maxsize=1)
+    await q.put({"message": "S1", "is_status_request": True})
+
+    # Test shutdown
+    setattr(q, "_is_shutdown", True)
+    shutdown_exc = getattr(asyncio, "QueueShutDown", Exception)
+    with pytest.raises(shutdown_exc):
+        await q.put({"message": "C1", "is_status_request": False})
+    setattr(q, "_is_shutdown", False)
+
+    # Test _wakeup_next skipping completed/cancelled waiters
+    waiter1: asyncio.Future[None] = asyncio.get_running_loop().create_future()
+    waiter1.cancel()  # already done/cancelled
+    waiter2: asyncio.Future[None] = asyncio.get_running_loop().create_future()
+
+    waiters: collections.deque[asyncio.Future[Any]] = collections.deque([waiter1, waiter2])
+    q._wakeup_next(waiters)
+
+    assert waiter1.cancelled()
+    assert waiter2.done()
+    assert waiter2.result() is None
+
+
 async def test_worker_sends_user_command_ahead_of_pending_status_polls(
     mock_handler: MagicMock,
 ) -> None:

From 55aa01de8a37df63ffcf345214c280948d385e38 Mon Sep 17 00:00:00 2001
From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com>
Date: Sat, 10 Oct 2026 12:25:48 +0200
Subject: [PATCH 4/4] docs(gateway_sessions): document envelope structure,
 shutdown parity, and storage complexity

---
 custom_components/myhome/gateway_sessions.py | 23 ++++++++++++++++++--
 1 file changed, 21 insertions(+), 2 deletions(-)

diff --git a/custom_components/myhome/gateway_sessions.py b/custom_components/myhome/gateway_sessions.py
index 97562e3d7..37d269d46 100644
--- a/custom_components/myhome/gateway_sessions.py
+++ b/custom_components/myhome/gateway_sessions.py
@@ -291,7 +291,15 @@ def close(self) -> None:
 
 
 def _classify_tier(item: Any) -> int:
-    """Classify item into priority tier: 0 (commands), 1 (status), 2 (others/sentinels)."""
+    """Classify item into priority tier: 0 (commands), 1 (status), 2 (others/sentinels).
+
+    Expected envelope dictionaries are produced by CommandWorkerPool._enqueue:
+    {
+        "message": OWNCommand,
+        "is_status_request": bool,
+        "written": asyncio.Future[float],
+    }
+    """
     if isinstance(item, dict):
         if item.get("is_status_request", False):
             return _TIER_STATUS
@@ -301,7 +309,14 @@ def _classify_tier(item: Any) -> int:
 
 
 class _PriorityQueueStorage:
-    """Multi-tier FIFO storage backing CommandPriorityQueue."""
+    """Multi-tier FIFO storage backing CommandPriorityQueue.
+
+    Implements the collections.abc.MutableSequence and deque interface expected
+    by queue inspection code and unit tests (e.g. list(handler.send_buffer._queue)).
+    Sequence materialisation (slices, __eq__, index) converts across tiers into
+    a list; with typical max_queue_size <= 250, this takes negligible CPU time (<2µs)
+    and avoids complex multi-deque virtual index bookkeeping.
+    """
 
     def __init__(self) -> None:
         """Initialize empty priority tiers."""
@@ -486,12 +501,16 @@ def _wakeup_next(self, waiters: collections.deque[asyncio.Future[Any]]) -> None:
     async def put(self, item: Any) -> None:
         """Put an item into the queue, prioritizing command waiters when full."""
         while self.full():
+            # Python 3.13+ asyncio.Queue.shutdown() sets `self._is_shutdown = True`
+            # and unblocks callers of put() with QueueShutDown. We mirror standard
+            # library semantics while preserving compatibility across Python versions.
             if getattr(self, "_is_shutdown", False):
                 raise getattr(asyncio, "QueueShutDown", Exception)
             putter: asyncio.Future[None] = asyncio.get_running_loop().create_future()
             is_cmd = _classify_tier(item) == _TIER_COMMANDS
             setattr(putter, "_is_cmd", is_cmd)
             if is_cmd:
+                # O(W) linear scan across waiting putters (W <= max_queue_size, typically 0 to <10).
                 idx = 0
                 for p in self._putters:
                     if getattr(p, "_is_cmd", False):