Match replies by osc/xid

.github/mutations/plan.json

@@ -152,5 +152,19 @@
     "find": "        self.coordinator_context = getattr(description, \"ssc_path\", None)",
     "replace": "        self.coordinator_context = None",
     "expect": "tests/test_integration_e2e.py::test_every_optional_path_has_an_entity_asking_for_it"
+  },
+  {
+    "name": "a late answer with another xid is taken as the answer",
+    "path": "custom_components/neumann_kh/ssc_client.py",
+    "find": "            if _is_late(reply_xid, xid):",
+    "replace": "            if False:",
+    "expect": "tests/test_ssc_client.py::test_a_line_with_another_xid_is_dropped"
+  },
+  {
+    "name": "a request over the token limit still carries an xid",
+    "path": "custom_components/neumann_kh/ssc_client.py",
+    "find": "        if \"osc\" in payload or json_tokens(payload) + _XID_TOKENS > _MAX_REQUEST_TOKENS:",
+    "replace": "        if \"osc\" in payload:",
+    "expect": "tests/test_ssc_client.py::test_a_request_over_the_token_limit_goes_without_an_xid"
   }
 ]

README.de.md

@@ -321,15 +321,15 @@ Entities sind „nicht verfügbar“, bis eine wieder geschlossen wird.
 - `ui/input_select` bleibt nur lesend. Der schreibbare Pfad ist
   `audio/in/interface`, bestätigt auf KH 120 II und KH 750 DSP und dort
   standardmäßig aktiviert.
-- Antwortet ein Lautsprecher auf eine einzelne Abfrage mit mehr als einer
-  Zeile, kann in einem engen Fall eine Folgezeile der einen Antwort als die
-  nächste ausgegeben werden. Die Integration sendet keine Transaktionskennung
-  (`osc/xid`, die beide Testgeräte zurückgeben), eine Folgezeile für den gerade
-  abgefragten Pfad ist von einer frischen Antwort darauf also nicht zu
-  unterscheiden. Bereits anliegende Zeilen werden
-  vor der nächsten Anfrage verworfen; betroffen ist nur eine, die genau
-  dazwischen eintrifft. Beide Testgeräte antworten mit exakt einer Zeile, das
-  betrifft also andere Firmware, nicht die eingesetzte Hardware.
+- Jede Wert-Anfrage trägt eine Transaktionskennung (`osc/xid`), die beide
+  Testgeräte in derselben Zeile zurückgeben, auch bei Fehlerantworten; eine
+  Zeile mit fremder Kennung ist eine verspätete Antwort auf eine frühere Anfrage
+  und wird verworfen. Ohne Kennung gehen nur EQ-Schreibvorgänge mit 20 Bändern,
+  weil die Kennung sie über die Größengrenze der Lautsprecher brächte, und die
+  Schema- und Limits-Abfrage der Geräte-Discovery; ihre Antworten werden über
+  den Pfad zugeordnet, wo in einem engen Fall eine Folgezeile der vorigen
+  Antwort als die nächste durchgehen könnte. Beide Testgeräte antworten mit
+  exakt einer Zeile.
 - Der Steuerungsmodus (`ui/control_mode`) wird mit dem Schalter CONTROL auf
   der Rückseite eingestellt (KH 750 DSP: CONTROL MODE) und lässt sich über das
   Netzwerk nicht ändern, deshalb ist er ein Sensor. Auf `LOCAL` antwortet der

README.md

@@ -308,14 +308,14 @@ unavailable until one is closed.
 - `ui/input_select` stays read-only. The writable path is
   `audio/in/interface`, confirmed on both the KH 120 II and the KH 750 DSP and
   enabled by default there.
-- A speaker that answers a single query with more than one line can, in a
-  narrow case, have a trailing line of one answer returned as the next one.
-  The integration sends no transaction id (`osc/xid`, which both test
-  speakers echo), so a trailing line for the path just requested cannot be
-  told apart from a fresh reply to it. Lines already waiting on the socket
-  are discarded before the next request; only one that
-  lands in between is affected. Both test devices answer with exactly one
-  line, so this concerns other firmware rather than the hardware in use.
+- Every value request carries a transaction id (`osc/xid`), which both test
+  speakers echo in the same line, error replies included; a line with another
+  id is a late answer to an earlier request and is dropped. Only EQ writes of
+  20 bands go without one, because the id would push them past the speakers'
+  request size limit, and the schema and limits walk of the device
+  discovery; their replies are matched by path, where a trailing line of the
+  previous answer could in a narrow case pass for the next one. Both test
+  devices answer with exactly one line.
 - Control mode (`ui/control_mode`) is set with the CONTROL switch on the
   back (KH 750 DSP: CONTROL MODE) and cannot be changed over the network, so
   it is a sensor. On `LOCAL` the speaker keeps answering, so its values stay

custom_components/neumann_kh/_util.py

@@ -37,6 +37,19 @@ def build_nested(path: tuple[str, ...], value: Any) -> dict[str, Any]:
     return root
 
 
+def json_tokens(value: Any) -> int:
+    """JSON tokens in a value: every object, array, key and scalar counts once.
+
+    The way a Neumann KH 120 II counts a request against its 29-token limit
+    (measured: a set of a 20-entry list four levels deep is exactly 29).
+    """
+    if isinstance(value, dict):
+        return 1 + sum(1 + json_tokens(child) for child in value.values())
+    if isinstance(value, list):
+        return 1 + sum(json_tokens(child) for child in value)
+    return 1
+
+
 def nesting_exceeds(value: Any, limit: int) -> bool:
     """Whether a parsed JSON value nests deeper than `limit`.
 

custom_components/neumann_kh/ssc_client.py

@@ -12,13 +12,14 @@
 
 import asyncio
 import contextlib
+import itertools
 import json
 import logging
 import os
 import socket
 from typing import Any
 
-from ._util import build_nested, deep_merge, extract, nesting_exceeds
+from ._util import build_nested, deep_merge, extract, json_tokens, nesting_exceeds
 from .const import DEFAULT_QUERY_SETTLE, DEFAULT_TIMEOUT
 
 _LOGGER = logging.getLogger(__name__)
@@ -67,6 +68,10 @@ def _os_reason(err: OSError) -> str:
 _MAX_LINE_BYTES = 1_048_576  # 1 MiB
 # Real replies nest a handful of levels (path plus osc/error); far past that is junk.
 _MAX_NESTING = 64
+# A KH 120 II refuses a request of more than 29 JSON tokens with 413, and
+# {"osc":{"xid":n}} adds 4: a 20-band EQ set fits only without it.
+_MAX_REQUEST_TOKENS = 29
+_XID_TOKENS = 4
 
 # Upper bounds for a single response. Every incoming line refreshes the settle
 # window, so a device that keeps talking would otherwise hold the read open
@@ -161,6 +166,45 @@ def _error_details(osc_error: Any) -> tuple[int | None, str]:
     return None, ""
 
 
+def _reply_xid(parsed: dict[str, Any]) -> Any:
+    osc = parsed.get("osc")
+    return osc.get("xid") if isinstance(osc, dict) else None
+
+
+def _is_late(reply_xid: Any, xid: int | None) -> bool:
+    """A line carrying another request's xid answers an earlier request."""
+    return reply_xid is not None and reply_xid != xid
+
+
+def _drop_xid(merged: dict[str, Any]) -> None:
+    """Callers walk the reply tree; the transaction id is the client's own business."""
+    osc = merged.get("osc")
+    if isinstance(osc, dict):
+        osc.pop("xid", None)
+        if not osc:
+            del merged["osc"]
+
+
+def _answer_complete(
+    merged: dict[str, Any], expect_path: tuple[str, ...] | None, reply_xid: Any
+) -> bool:
+    """The answer ends with an error or the requested value.
+
+    The own xid alone ends it only where no path was asked for: a firmware
+    that sends the xid before the value would otherwise leave the value line
+    behind, to be taken as the answer to the next request.
+    """
+    if extract(merged, ("osc", "error")) is not None:
+        return True
+    if expect_path is None:
+        return reply_xid is not None
+    if reply_xid is not None:
+        # Matched by xid, so a JSON null there is the answer, not a gap.
+        parent = extract(merged, expect_path[:-1])
+        return isinstance(parent, dict) and expect_path[-1] in parent
+    return extract(merged, expect_path) is not None
+
+
 def _parse_reply(line: bytes) -> dict[str, Any] | None:
     """A reply line as an address tree, or None if the line is unusable."""
     try:
@@ -202,6 +246,11 @@ def __init__(
         # Tells the poll loop a user action is waiting (see request(priority=
         # True)); the poll then pauses before its next path.
         self._priority_waiting = asyncio.Event()
+        self._xids = itertools.count(1)
+        # Whether the last answer on this connection was matched by path only.
+        # Only then can a trailing line of it pass for the next answer, so
+        # only then is the stale drain worth its wait.
+        self._matched_by_path = False
         # Logical path -> path on this device (see use_path_aliases).
         self._path_aliases: dict[tuple[str, ...], tuple[str, ...]] = {}
 
@@ -255,6 +304,7 @@ async def _ensure_connected(self) -> bool:
 
     def _drop_connection(self) -> None:
         """Drops the current connection; the next access reconnects."""
+        self._matched_by_path = False
         if self._writer is not None:
             self._writer.close()
         self._writer = None
@@ -350,7 +400,7 @@ async def _discard_stale_lines(self) -> None:
         self._drop_connection()
 
     async def _read_lines_until_settled(
-        self, expect_path: tuple[str, ...] | None = None
+        self, expect_path: tuple[str, ...] | None = None, xid: int | None = None
     ) -> dict[str, Any]:
         """Read response lines and merge them into one result dict.
 
@@ -363,24 +413,19 @@ async def _read_lines_until_settled(
         The settle window still applies while the answer is incomplete, so a
         firmware that splits a response across lines is still handled.
 
-        Known limitation, accepted deliberately. Returning early can leave a
-        trailing line of that answer in flight. _discard_stale_lines() catches
-        one that is already queued when the next request goes out, but not one
-        that lands after the drain and before the real reply - and since the
-        client sends no transaction id (the firmware would echo osc/xid), a
-        trailing line for the requested path is indistinguishable from a fresh
-        answer to it.
-
-        There is no cheap fix: waiting out the settle window every time is
-        what the early return exists to avoid, reconnecting per request costs
-        more, and learning from what the drain found does not help because in
-        exactly this case the drain sees nothing. Measured against both test
-        devices a leaf query is answered with exactly one line, so this is a
-        risk for other firmware rather than for the hardware in use.
+        A line carrying another request's osc/xid is a late answer and is
+        dropped; the line with this request's xid is the whole answer (the
+        device echoes it in the same line, error replies included). A request
+        without an xid - osc methods, a set too long to carry one - is matched
+        by path. That keeps the known limitation for those few: a trailing
+        line of the previous answer that lands after the stale drain is
+        indistinguishable from a fresh answer to the same path.
         """
         if self._reader is None:
             raise SSCConnectionError(f"No active connection to {mask_host(self._host)}")
         merged: dict[str, Any] = {}
+        # Until a line echoes the xid, this answer counts as matched by path.
+        self._matched_by_path = True
         received = False
         limit_hit = False
         lines = 0
@@ -451,11 +496,14 @@ async def _read_lines_until_settled(
                 )
                 continue
 
+            reply_xid = _reply_xid(parsed)
+            if _is_late(reply_xid, xid):
+                _LOGGER.debug("Dropped a late answer to an earlier request from %s", self._host)
+                continue
             received = True
             deep_merge(merged, parsed)
-            if extract(merged, ("osc", "error")) is not None:
-                break
-            if expect_path is not None and extract(merged, expect_path) is not None:
+            if _answer_complete(merged, expect_path, reply_xid):
+                self._matched_by_path = reply_xid is None
                 break
 
             # Both bounds are evaluated at the top of the next pass, which is
@@ -505,11 +553,13 @@ async def _locked_request(
             # await, so both can be cancelled. Left outside, a cancellation
             # hitting that window would keep a connection the caller believes
             # was dropped - the invariant the v1.15.0 hardening relies on.
-            # Nothing can be stale on a socket opened for this request.
-            if not await self._ensure_connected():
+            xid = self._next_xid(payload)
+            # Nothing can be stale on a socket opened for this request, and
+            # a late line of an answer matched by xid carries its own id.
+            if not await self._ensure_connected() and self._matched_by_path:
                 await self._discard_stale_lines()
-            await self._send_raw(payload)
-            merged = await self._read_lines_until_settled(expect_path)
+            await self._send_raw(payload if xid is None else {"osc": {"xid": xid}, **payload})
+            merged = await self._read_lines_until_settled(expect_path, xid)
         except SSCConnectionError, SSCTimeoutError:
             # Drop the connection so the next attempt reconnects
             self._drop_connection()
@@ -522,6 +572,7 @@ async def _locked_request(
             self._drop_connection()
             raise
 
+        _drop_xid(merged)
         osc_error = extract(merged, ("osc", "error"))
         if osc_error:
             code, description = _error_details(osc_error)
@@ -533,6 +584,16 @@ async def _locked_request(
 
         return merged
 
+    def _next_xid(self, payload: dict[str, Any]) -> int | None:
+        """An id for a value request, or None where the device would not take one.
+
+        osc methods were never measured with an xid, and a request the xid
+        would push past the token limit is refused (413) - both go without.
+        """
+        if "osc" in payload or json_tokens(payload) + _XID_TOKENS > _MAX_REQUEST_TOKENS:
+            return None
+        return next(self._xids)
+
     def use_path_aliases(self, aliases: dict[tuple[str, ...], tuple[str, ...]]) -> None:
         """Send these logical paths to a different path on this device.
 

tests/README.md

@@ -100,9 +100,9 @@ headroom for the next person to spend.
 
 The measure is Cognitive Complexity (Campbell / SonarSource), implemented in
 `complexity.py` and calibrated against SonarQube Cloud: at calibration Sonar
-reported 35 for `_read_lines_until_settled`, and so did this. (Today it is 33,
-after the reply parsing moved out into `_parse_reply`.) `COMPLEXITY_LIMIT` is 15,
-SonarSource's default.
+reported 35 for `_read_lines_until_settled`, and so did this. It is lower
+today, since parts moved into helpers; `test_budgets.py` holds the current
+value. `COMPLEXITY_LIMIT` is 15, SonarSource's default.
 
 ## Adding a guard
 

tests/complexity.py

@@ -3,7 +3,8 @@
 A measuring stick for the budget guard next door, not a general-purpose
 implementation. Calibrated against SonarQube Cloud during the complexity
 round of 1.18.1: Sonar reported 35 for `_read_lines_until_settled`, and this
-scorer returned 35 for it then (33 since `_parse_reply` was split out).
+scorer returned 35 for it then (lower since parts moved into helpers;
+test_budgets.py holds the current value).
 
 Three rules:
 

tests/recorded_replies.py

@@ -129,6 +129,13 @@
     '{"osc":{"error":[{"audio":{"out":{"mute":[401,{"desc":"authorisation needed"}]}}}]}}',
 )
 
+# On LOCAL a read-only leaf keeps its 405: writability is checked before the
+# switch (KH 120 II, 2026-11-01, sent with an xid; the xid is left out here).
+LOCAL_MODE_READ_ONLY_SET_REPLY = (
+    {"ui": {"control_mode": "LOCAL"}},
+    '{"osc":{"error":[{"ui":{"control_mode":[405,{"desc":"method not allowed"}]}}]}}',
+)
+
 # osc/schema and osc/limits - answered by both models in exactly the form the
 # integration sends (an older note said 400; it is not).
 SCHEMA_ROOT_REPLY = (
@@ -164,6 +171,42 @@
     ),
 )
 
+# Error replies to requests with an xid, both models alike (2026-11-01): the
+# xid sits next to "error" in the same osc object, never in a second one.
+XID_ERROR_REPLIES = (
+    (
+        {"osc": {"xid": 101}, "audio": {"out": {"dimm": None}}},
+        '{"osc":{"error":[{"audio":{"out":{"dimm":[404,{"desc":"address not found"}]}}}],"xid":101}}',
+    ),
+    (
+        {"osc": {"xid": 102}, "ui": {"control_mode": "NETWORK"}},
+        '{"osc":{"error":[{"ui":{"control_mode":[405,{"desc":"method not allowed"}]}}],"xid":102}}',
+    ),
+    (
+        {"osc": {"xid": 103}, "audio": {"out": {"mute": "x"}}},
+        '{"osc":{"error":[{"audio":{"out":{"mute":[406,{"desc":"not acceptable"}]}}}],"xid":103}}',
+    ),
+    (
+        {"osc": {"xid": 104}, "audio": {"out": {"level": None, "dimm": None}}},
+        (
+            '{"audio":{"out":{"level":100.0}},'
+            '"osc":{"error":[{"audio":{"out":[404,{"desc":"address not found"}]}}],"xid":104}}'
+        ),
+    ),
+    (
+        {"osc": {"xid": 105, "nonexistent": None}},
+        '{"osc":{"error":[{"osc":[404,{"desc":"address not found"}]}],"xid":105}}',
+    ),
+)
+
+# More than 29 JSON tokens, both models (KH 120 II 2026-10-20, KH 750
+# 2026-11-01): refused before anything is read, so the reply carries no xid.
+# Here a get bundle of 13 unknown leaves (31 tokens); 12 (29 tokens) answer 404.
+REQUEST_TOO_LONG_REPLY = (
+    {"osc": {"xid": 1}, "audio": {"out": {f"x{n}": None for n in range(13)}}},
+    '{"osc":{"error":[413,{"desc":"request too long"}]}}',
+)
+
 # Protocol methods of the osc container and the transaction id, both models.
 OSC_METHOD_REPLIES = (
     ({"osc": {"version": None}}, '{"osc":{"version":"1.1"}}'),

tests/test_budgets.py

@@ -42,7 +42,9 @@
     # Raised to 556 for _parse_reply: the nesting limit on faulty replies, taken
     # out of the read loop so its complexity went down instead of up.
     # Raised to 559 to skip the stale drain on a connection opened just now.
-    "ssc_client.py": 559,
+    # Raised to 608 for matching replies by osc/xid (token limit, late lines).
+    # Raised to 620 so a split xid answer waits for its value (re-audit R1).
+    "ssc_client.py": 620,
 }
 
 # SonarSource's default. Everything above needs a written entry.
@@ -51,7 +53,7 @@
     # The one function this project keeps having trouble with. Kept whole by
     # owner decision (ledger A8, 2026-09-02) rather than split under pressure:
     # every previous attempt moved the branching instead of removing it.
-    "ssc_client.py::SSCClient._read_lines_until_settled": 33,
+    "ssc_client.py::SSCClient._read_lines_until_settled": 32,
     # Writes, counts and reports per path, with a deadline between paths and
     # three ways out that all have to keep what already landed (the third:
     # LOCAL mode, where every further write would be refused too).