diff --git a/custom-providers/xiaozhi-patches/http_server.py b/custom-providers/xiaozhi-patches/http_server.py
index f403eb22..d66165bf 100644
--- a/custom-providers/xiaozhi-patches/http_server.py
+++ b/custom-providers/xiaozhi-patches/http_server.py
@@ -438,6 +438,20 @@ async def _dotty_play_asset(self, request: "web.Request") -> "web.Response":
             )
         resolved_id = (getattr(conn, "headers", {}) or {}).get("device-id", "") or device_id
 
+        # DOTTY-PATCH: don't clobber a live chat turn. play-asset is timer-driven
+        # on the first available device, so a purr/song landing mid-conversation
+        # would reset client_abort/client_is_speaking (below) and mark the device
+        # idle in its finally block while chat TTS is still streaming — cancelling
+        # the user's barge-in and corrupting the turn. Refuse instead; a dropped
+        # ambient asset is the correct outcome here.
+        if getattr(conn, "client_is_speaking", False) and not getattr(
+            conn, "client_abort", False
+        ):
+            return web.json_response(
+                {"error": "device busy (mid-turn)", "device_id": resolved_id},
+                status=409,
+            )
+
         async def _dispatch() -> None:
             import json as _json
             import numpy as _np
diff --git a/custom-providers/xiaozhi-patches/ota_handler.py b/custom-providers/xiaozhi-patches/ota_handler.py
index c5545765..d1896c88 100644
--- a/custom-providers/xiaozhi-patches/ota_handler.py
+++ b/custom-providers/xiaozhi-patches/ota_handler.py
@@ -21,26 +21,35 @@ def _safe_basename(filename: str) -> str:
     return os.path.basename(filename)
 
 
-def _parse_version(ver: str) -> Tuple[int, ...]:
-    # conservative parser: split by non-digit, keep numeric parts
-    parts = re.findall(r"\d+", ver)
-    return tuple(int(p) for p in parts) if parts else (0,)
+def _parse_version(ver: str) -> Tuple:
+    """Comparison/sort key for a firmware version string.
+
+    Considers only the first three numeric segments (major.minor.patch). A
+    trailing pre-release / build suffix (anything after the numeric core, e.g.
+    ``-rc1``, ``-beta``, ``+build5``) ranks BELOW the corresponding release,
+    matching semver precedence — so a device on GA ``1.2.3`` is never offered a
+    stale ``1.2.3-rc1``. The naive ``re.findall(r"\\d+")`` parser this replaced
+    turned ``1.2.3-rc1`` into ``(1,2,3,1)`` and ranked it ABOVE GA.
+
+    The returned tuple is shaped ``(major, minor, patch, release_rank, suffix)``
+    where release_rank is 1 for a clean release and 0 for any suffixed build,
+    and suffix is compared lexically to order pre-releases deterministically
+    (``-alpha`` < ``-beta``). Element types are positionally consistent, so
+    these keys are safe to compare and sort.
+    """
+    s = ver.strip().lstrip("vV")
+    m = re.match(r"\d+(?:\.\d+)*", s)
+    core = m.group(0) if m else ""
+    nums = [int(p) for p in core.split(".") if p != ""]
+    nums = (nums + [0, 0, 0])[:3]
+    suffix = s[len(core):].lstrip(".-_+ ").strip()
+    release_rank = 1 if suffix == "" else 0
+    return (nums[0], nums[1], nums[2], release_rank, suffix)
 
 
 def _is_higher_version(a: str, b: str) -> bool:
-    """Return True if version string a > b (semver-like numeric compare)."""
-    ta = _parse_version(a)
-    tb = _parse_version(b)
-    # compare tuple lexicographically, but allow different lengths
-    maxlen = max(len(ta), len(tb))
-    for i in range(maxlen):
-        ai = ta[i] if i < len(ta) else 0
-        bi = tb[i] if i < len(tb) else 0
-        if ai > bi:
-            return True
-        if ai < bi:
-            return False
-    return False
+    """Return True if version string a > b (semver-like precedence)."""
+    return _parse_version(a) > _parse_version(b)
 
 
 class OTAHandler(BaseHandler):
diff --git a/tests/test_ota_version.py b/tests/test_ota_version.py
new file mode 100644
index 00000000..24642398
--- /dev/null
+++ b/tests/test_ota_version.py
@@ -0,0 +1,101 @@
+"""Unit tests for the OTA version comparator (ota_handler.py).
+
+Covers the pre-release / build-suffix precedence bug: the old
+``re.findall(r"\\d+")`` parser turned ``1.2.3-rc1`` into ``(1,2,3,1)`` and
+ranked it ABOVE GA ``1.2.3``, so a device on GA would be offered a stale rc.
+The module's core.* / aiohttp imports are stubbed so the test runs without a
+container; only the pure module-level functions are exercised.
+"""
+import pathlib
+import sys
+import types
+import unittest
+from unittest.mock import MagicMock
+
+# ── Stub the container-only imports ota_handler pulls at load time ────────────
+# NB: do NOT stub `aiohttp` — it's a real installed dep that other test modules
+# in this suite import for real; stubbing it here (setdefault wins on import
+# order) would poison sys.modules and break them.
+for _n in ("core", "core.auth", "core.utils", "core.utils.util", "core.api"):
+    sys.modules.setdefault(_n, MagicMock())
+
+# BaseHandler is used as a real base class (`class OTAHandler(BaseHandler)`), so
+# it must be a genuine type, not a MagicMock attribute.
+_base_mod = types.ModuleType("core.api.base_handler")
+
+
+class _StubBase:
+    def __init__(self, *a, **k):
+        pass
+
+
+_base_mod.BaseHandler = _StubBase  # type: ignore[attr-defined]
+sys.modules["core.api.base_handler"] = _base_mod
+
+# ── Load the module under test by path ───────────────────────────────────────
+import importlib.util as _ilu
+
+_OTA_PY = (
+    pathlib.Path(__file__).parent.parent
+    / "custom-providers"
+    / "xiaozhi-patches"
+    / "ota_handler.py"
+)
+_spec = _ilu.spec_from_file_location("ota_handler_under_test", _OTA_PY)
+_mod = _ilu.module_from_spec(_spec)  # type: ignore[arg-type]
+_spec.loader.exec_module(_mod)  # type: ignore[union-attr]
+
+_is_higher = _mod._is_higher_version
+_key = _mod._parse_version
+
+
+class TestOtaVersion(unittest.TestCase):
+
+    # The headline regression: a pre-release must NOT outrank its GA release.
+    def test_prerelease_not_higher_than_release(self):
+        self.assertFalse(_is_higher("1.2.3-rc1", "1.2.3"))
+        self.assertTrue(_is_higher("1.2.3", "1.2.3-rc1"))
+
+    def test_build_suffix_not_higher_than_release(self):
+        self.assertFalse(_is_higher("1.2.3+build5", "1.2.3"))
+        self.assertTrue(_is_higher("1.2.3", "1.2.3+build5"))
+
+    def test_normal_ordering(self):
+        self.assertTrue(_is_higher("1.2.4", "1.2.3"))
+        self.assertTrue(_is_higher("2.0.0", "1.9.9"))
+        self.assertTrue(_is_higher("1.3.0", "1.2.9"))
+        self.assertFalse(_is_higher("1.2.3", "1.2.4"))
+
+    def test_equal_versions_are_not_higher(self):
+        self.assertFalse(_is_higher("1.2.3", "1.2.3"))
+        # short form equals its zero-padded form (1.2 == 1.2.0)
+        self.assertFalse(_is_higher("1.2", "1.2.0"))
+        self.assertFalse(_is_higher("1.2.0", "1.2"))
+        self.assertEqual(_key("1.2"), _key("1.2.0"))
+
+    def test_leading_v_stripped(self):
+        self.assertFalse(_is_higher("v1.2.3", "1.2.3"))
+        self.assertTrue(_is_higher("v1.2.4", "v1.2.3"))
+
+    def test_prerelease_ordering_is_deterministic(self):
+        # Among pre-releases of the same core, lexical suffix order applies.
+        self.assertTrue(_is_higher("1.0.0-beta", "1.0.0-alpha"))
+        self.assertFalse(_is_higher("1.0.0-alpha", "1.0.0-beta"))
+
+    def test_only_first_three_segments_considered(self):
+        # A 4th numeric segment must not flip precedence vs the 3-segment GA.
+        self.assertFalse(_is_higher("1.2.3.4", "1.2.3"))
+        self.assertEqual(_key("1.2.3.4")[:3], (1, 2, 3))
+
+    def test_key_is_sortable(self):
+        versions = ["1.2.3", "1.2.3-rc1", "1.10.0", "1.2.10", "2.0.0", "v1.2.4"]
+        ordered = sorted(versions, key=_key)
+        # GA 1.2.3 sorts above its rc; numeric (not lexical) segment compare.
+        self.assertLess(ordered.index("1.2.3-rc1"), ordered.index("1.2.3"))
+        self.assertLess(ordered.index("1.2.3"), ordered.index("1.2.10"))
+        self.assertLess(ordered.index("1.2.10"), ordered.index("1.10.0"))
+        self.assertEqual(ordered[-1], "2.0.0")
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_play_asset_route.py b/tests/test_play_asset_route.py
index 701d6cec..81080c83 100644
--- a/tests/test_play_asset_route.py
+++ b/tests/test_play_asset_route.py
@@ -164,6 +164,20 @@ def test_no_device_returns_503(self):
         self.assertEqual(resp.status, 503)
         self.assertIn("known", _body(resp))
 
+    def test_mid_turn_device_returns_409(self):
+        # Device actively speaking (chat TTS streaming) and not aborted → a
+        # timer-driven asset must not clobber the turn; refuse with 409.
+        path = self._asset()
+        conn = MagicMock()
+        conn.headers = {"device-id": "dev1"}
+        conn.client_is_speaking = True
+        conn.client_abort = False
+        _mock_active["dev1"] = conn
+        req = _FakeRequest(data={"asset": path, "device_id": "dev1"})
+        resp = _run(self.srv._dotty_play_asset(req))
+        self.assertEqual(resp.status, 409)
+        self.assertIn("busy", _body(resp)["error"])
+
     # Happy-path ───────────────────────────────────────────────────────────────
 
     def test_valid_request_fires_task_and_returns_200(self):