diff --git a/CHANGELOG.md b/CHANGELOG.md
index f994359..89366a7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## [Unreleased]
 
+### Added
+- **Voice model pre-warm on connect** (#181): when a device WebSocket opens, PiVoiceLLM sends llama-swap a one-token request for `qwen3.5:4b` (at most once per 5 minutes, off the connection's path, failures only logged) so a model that was unloaded is loading while the person is still speaking, not after. URL comes from `LLM.PiVoiceLLM.prewarm_url` / `DOTTY_PREWARM_URL`, else the `VISION_BRIDGE_URL` host on :8080; `prewarm: false` turns it off.
+
+### Fixed
+- **Room noise closes the session** (#105): after 3 noise-like ASR turns in a row (empty or dropped transcripts, `no_speech` >= 0.45 or `avg_logprob` < -1.0, each within 90 s of the last), WhisperLocal ends the listening session as the no-voice timeout does; touch or the wake word reopens it. A real turn resets the run. Config keys under `ASR.WhisperLocal`: `noise_turn_limit` (default 3, `0` disables), `noise_no_speech`, `noise_avg_logprob`. Clearly articulated TV speech still passes these signals; only a wake word or voiceprint can separate it. The threshold of 3 is awaiting the maintainer's confirmation.
+- **Latest utterance wins** (#105): a turn queued behind the #175 turn lock is skipped when the same session said something newer while it waited.
+
 ## [fw-v1.3.5] - 2026-10-29
 
 Firmware only (`BrettKinny/StackChan@dotty`, submodule `c074907`). Source-only: the server address is compiled in, so build it yourself with your own `CONFIG_OTA_URL` (SETUP.md §2). Drafted with AI assistance (Claude Opus 5.5) and smoke-tested on a physical StackChan.
diff --git a/custom-providers/asr/whisper_local.py b/custom-providers/asr/whisper_local.py
index 89c75bc..4f43a12 100644
--- a/custom-providers/asr/whisper_local.py
+++ b/custom-providers/asr/whisper_local.py
@@ -32,6 +32,94 @@ def _no_speech_threshold(config: dict) -> float:
     return value if 0.0 <= value <= 1.0 else DEFAULT_NO_SPEECH_THRESHOLD
 
 
+# #105: room noise (TV, podcast, a busy kitchen) keeps a session open and feeds
+# Whisper fake turns. The maintainer's direction is to close the session after
+# a run of noise-like turns and let touch wake it. Level alone cannot tell this
+# noise from a quiet child (rms 0.002-0.014 vs 0.007-0.017), so the signals are
+# Whisper's own confidence. Real close-talk on the robot scored no_speech <=
+# 0.32 and avg_logprob >= -0.9; the room-noise run scored no_speech up to 0.59.
+DEFAULT_NOISE_TURN_LIMIT = 3  # 0 disables; Brett to confirm
+DEFAULT_NOISE_NO_SPEECH = 0.45
+DEFAULT_NOISE_LOGPROB = -1.0  # Whisper's own "failed decode" line
+# Noise turns further apart than this are separate episodes, not a run.
+NOISE_RUN_WINDOW_S = 90.0
+
+
+def _float_option(config: dict, key: str, default: float) -> float:
+    try:
+        return float(config.get(key, default))
+    except (TypeError, ValueError):
+        return default
+
+
+def _int_option(config: dict, key: str, default: int) -> int:
+    try:
+        return max(0, int(config.get(key, default)))
+    except (TypeError, ValueError):
+        return default
+
+
+def _is_noise_like(content: str, no_speech: float, avg_logprob: float,
+                   no_speech_cutoff: float, logprob_cutoff: float) -> bool:
+    """A turn that is probably not someone talking to Dotty.
+
+    An empty transcript counts: VAD opened on sound that Whisper (or the
+    no-speech gate) would not turn into words. Otherwise the transcript needs
+    to be a weak decode or likely non-speech."""
+    if not content.strip():
+        return True
+    return no_speech >= no_speech_cutoff or avg_logprob < logprob_cutoff
+
+
+class NoiseRunTracker:
+    """Counts consecutive noise-like turns per session."""
+
+    def __init__(self, limit: int, window_s: float = NOISE_RUN_WINDOW_S):
+        self.limit = limit
+        self.window_s = window_s
+        self._runs: dict = {}  # session_id -> (count, last_time)
+
+    def record(self, session_id, noise_like: bool, now: float) -> bool:
+        """Note one turn. True when the run has reached the limit (the run is
+        then reset so the caller closes the session once)."""
+        if self.limit <= 0:
+            return False
+        if not noise_like:
+            self._runs.pop(session_id, None)
+            return False
+        count, last = self._runs.get(session_id, (0, now))
+        if now - last > self.window_s:
+            count = 0
+        count += 1
+        if count >= self.limit:
+            self._runs.pop(session_id, None)
+            return True
+        self._runs[session_id] = (count, now)
+        # Forget long-dead sessions so the map cannot grow without bound.
+        if len(self._runs) > 64:
+            self._runs = {k: v for k, v in self._runs.items()
+                          if now - v[1] <= self.window_s}
+        return False
+
+
+async def _close_session_for_noise(session_id, count: int) -> None:
+    """End the device session (touch/wake word reopens it). Never raises."""
+    try:
+        from core.portal_bridge import active_connections
+        from core.utils import device_command
+
+        conn = next((c for c in list(active_connections.values())
+                     if getattr(c, "session_id", None) == session_id), None)
+        if conn is None:
+            logger.bind(tag=TAG).warning(
+                f"ASR-NOISE-CLOSE no live connection for session {session_id!r}")
+            return
+        await device_command.end_session(
+            conn, reason=f"{count} noise-like turns in a row (#105)")
+    except Exception as e:
+        logger.bind(tag=TAG).warning(f"ASR-NOISE-CLOSE failed (non-fatal): {e}")
+
+
 def _hotwords(config: dict) -> Optional[str]:
     """Decoder bias for rare words such as the wake-name (#164). None = off."""
     value = config.get("hotwords")
@@ -90,6 +178,12 @@ def __init__(self, config: dict, delete_audio_file: bool):
         self.initial_prompt = config.get("initial_prompt", None)
         self.hotwords = _hotwords(config)
         self.no_speech_threshold = _no_speech_threshold(config)
+        self.noise_no_speech = _float_option(
+            config, "noise_no_speech", DEFAULT_NOISE_NO_SPEECH)
+        self.noise_logprob = _float_option(
+            config, "noise_avg_logprob", DEFAULT_NOISE_LOGPROB)
+        self.noise_runs = NoiseRunTracker(
+            _int_option(config, "noise_turn_limit", DEFAULT_NOISE_TURN_LIMIT))
         self.delete_audio_file = delete_audio_file
 
         if self.output_dir:
@@ -213,6 +307,10 @@ async def speech_to_text(
                 except Exception as _e:
                     logger.bind(tag=TAG).warning(f"ASR-METRICS log failed (non-fatal): {_e}")
 
+                if self._note_turn_for_noise(session_id, content, segments):
+                    # The run's last noise turn is not answered either.
+                    text = {"content": ""}
+
                 return text, artifacts.file_path
 
             except OSError as e:
@@ -233,6 +331,29 @@ async def speech_to_text(
 
         return "", None
 
+    def _note_turn_for_noise(self, session_id, content: str, segments) -> bool:
+        """#105: close the session after a run of noise-like turns. Never
+        raises: this must not cost a real turn. True when the session is
+        being closed."""
+        try:
+            if segments:
+                nsp = sum(s.no_speech_prob for s in segments) / len(segments)
+                alp = sum(s.avg_logprob for s in segments) / len(segments)
+            else:
+                nsp, alp = 1.0, 0.0
+            noise = _is_noise_like(
+                content, nsp, alp, self.noise_no_speech, self.noise_logprob)
+            if self.noise_runs.record(session_id, noise, time.monotonic()):
+                logger.bind(tag=TAG).info(
+                    f"ASR-NOISE-CLOSE {self.noise_runs.limit} noise-like turns "
+                    f"in a row; ending session {session_id!r}")
+                asyncio.get_running_loop().create_task(
+                    _close_session_for_noise(session_id, self.noise_runs.limit))
+                return True
+        except Exception as e:
+            logger.bind(tag=TAG).warning(f"noise tracking failed (non-fatal): {e}")
+        return False
+
     def _transcribe_blocking(self, audio: np.ndarray):
         """Run inside asyncio.to_thread. Eagerly consume the segment iterator
         here so the async caller doesn't iterate a generator backed by C++
diff --git a/custom-providers/pi_voice/pi_voice.py b/custom-providers/pi_voice/pi_voice.py
index 2ec8ed8..38ce0de 100644
--- a/custom-providers/pi_voice/pi_voice.py
+++ b/custom-providers/pi_voice/pi_voice.py
@@ -173,6 +173,31 @@ def _dashboard_url() -> str:
     return f"{parsed.scheme or 'http'}://{parsed.hostname}:{_DASHBOARD_PORT}"
 
 
+_LLM_PORT = 8080
+_DEFAULT_PREWARM_MODEL = "qwen3.5:4b"
+_PREWARM_MIN_INTERVAL_S = 300.0
+
+
+def _prewarm_url(config: dict) -> str:
+    """Chat-completions URL used to wake the voice model, or "" when off.
+
+    Config key `prewarm_url` / env `DOTTY_PREWARM_URL` win; `prewarm: false`
+    switches it off. Otherwise llama-swap is assumed to sit on the sibling
+    host's :8080 (where dotty-pi reaches it)."""
+    if str(config.get("prewarm", "true")).strip().lower() in ("0", "false", "no", "off"):
+        return ""
+    explicit = str(
+        config.get("prewarm_url") or os.environ.get("DOTTY_PREWARM_URL", "")
+    ).strip()
+    if explicit:
+        return explicit
+    sibling = os.environ.get("BRIDGE_URL") or os.environ.get("VISION_BRIDGE_URL", "")
+    parsed = urlparse(sibling)
+    if not parsed.hostname:
+        return ""
+    return f"{parsed.scheme or 'http'}://{parsed.hostname}:{_LLM_PORT}/v1/chat/completions"
+
+
 def _report_to_dashboard(path: str, payload: dict) -> None:
     """Fire-and-forget POST to the bridge dashboard.
 
@@ -211,6 +236,10 @@ def _last_user_text(dialogue: list[dict]) -> str:
     return ""
 
 
+def _normalise_user_text_for_claim(dialogue: list[dict]) -> str:
+    return split_room_view(_last_user_text(dialogue))[1].strip()
+
+
 def _normalise_user_content(content: object) -> str:
     """Extract xiaozhi's user text without leaking its JSON envelope to pi.
 
@@ -341,18 +370,97 @@ def __init__(
         # agent_end transaction exclusive; per-write locking cannot prevent one
         # caller from consuming another caller's response frames.
         self._turn_lock = threading.Lock()
+        # Latest utterance wins (#105): a turn that waited behind the lock is
+        # dropped if the same session said something newer meanwhile, instead
+        # of being answered late and out of date.
+        self._turn_seq_lock = threading.Lock()
+        self._turn_seq = 0
+        self._latest_turn: dict[object, int] = {}
         self._first_turn = True
+        self._prewarm_url = _prewarm_url(config)
+        self._prewarm_model = str(config.get("prewarm_model") or _DEFAULT_PREWARM_MODEL)
+        self._last_prewarm = float("-inf")
         msg = f"PiVoiceLLM ready (container={self._container} kid_mode={self._kid_mode})"
         try:
             logger.bind(tag=TAG).info(msg)  # type: ignore[attr-defined]
         except AttributeError:
             logger.info(msg)
 
+    def prewarm(self) -> bool:
+        """Wake the voice model before the first turn (#181).
+
+        llama-swap can unload the 4B between conversations and the first
+        reply then takes ~12 s, with the mic open the whole time. The device
+        WebSocket opening (wake word, touch, face) is the earliest signal, so
+        the server calls this then. Fire-and-forget on a daemon thread; at
+        most once per few minutes. Returns True when a warm-up was started."""
+        if not self._prewarm_url:
+            return False
+        now = time.monotonic()
+        with self._turn_seq_lock:
+            if now - self._last_prewarm < _PREWARM_MIN_INTERVAL_S:
+                return False
+            self._last_prewarm = now
+        url, model = self._prewarm_url, self._prewarm_model
+
+        def _post() -> None:
+            started = time.monotonic()
+            try:
+                req = urllib.request.Request(
+                    url,
+                    data=json.dumps({
+                        "model": model,
+                        "messages": [{"role": "user", "content": "hi"}],
+                        "max_tokens": 1,
+                    }).encode("utf-8"),
+                    headers={"Content-Type": "application/json"},
+                    method="POST",
+                )
+                urllib.request.urlopen(req, timeout=90).close()
+                logger.info(f"PiVoiceLLM prewarm ok in {time.monotonic() - started:.1f}s")
+            except Exception as exc:
+                logger.warning(f"PiVoiceLLM prewarm failed (non-fatal): {exc}")
+
+        threading.Thread(target=_post, name="dotty-prewarm", daemon=True).start()
+        return True
+
     # xiaozhi-server's voice loop calls this as a sync generator.
     # Each yielded string becomes a TTS chunk.
     def response(self, session_id, dialogue, **kwargs) -> Iterator[str]:
-        with self._turn_lock:
-            yield from self._response_serialized(session_id, dialogue, **kwargs)
+        seq = self._claim_turn(session_id, dialogue)
+        try:
+            with self._turn_lock:
+                if seq is not None and self._is_superseded(session_id, seq):
+                    logger.info(
+                        "PiVoiceLLM: skipping queued turn; a newer utterance "
+                        "arrived while it waited (#105)"
+                    )
+                    return
+                yield from self._response_serialized(session_id, dialogue, **kwargs)
+        finally:
+            self._release_turn(session_id, seq)
+
+    def _claim_turn(self, session_id, dialogue) -> int | None:
+        """Number this turn before it waits for the lock. Turns with no user
+        words are not numbered: they must not supersede a real one."""
+        if not _normalise_user_text_for_claim(dialogue):
+            return None
+        with self._turn_seq_lock:
+            self._turn_seq += 1
+            self._latest_turn[session_id] = self._turn_seq
+            return self._turn_seq
+
+    def _is_superseded(self, session_id, seq: int) -> bool:
+        with self._turn_seq_lock:
+            return self._latest_turn.get(session_id, seq) != seq
+
+    def _release_turn(self, session_id, seq: int | None) -> None:
+        # Forget the session once its newest turn is done so the map cannot grow.
+        if seq is None:
+            return
+        with self._turn_seq_lock:
+            if self._latest_turn.get(session_id) == seq:
+                del self._latest_turn[session_id]
 
     def _response_serialized(self, session_id, dialogue, **kwargs) -> Iterator[str]:
         """Run one complete Pi RPC transaction while ``_turn_lock`` is held."""
diff --git a/custom-providers/pi_voice/tests/test_pi_voice.py b/custom-providers/pi_voice/tests/test_pi_voice.py
index be6da17..4526eb3 100644
--- a/custom-providers/pi_voice/tests/test_pi_voice.py
+++ b/custom-providers/pi_voice/tests/test_pi_voice.py
@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 
+import json
 import os
 import sys
 import tempfile
@@ -213,6 +214,132 @@ def test_no_user_message_short_circuits(self):
         self.assertEqual(client.prompts, [], "PiClient must not be called for empty dialogue")
 
 
+class TestLatestUtteranceWins(unittest.TestCase):
+    """#105: a turn queued behind the turn lock is dropped when the same
+    session said something newer while it waited."""
+
+    def _run(self, texts_by_session):
+        class GatedClient(FakeClient):
+            def __init__(self):
+                super().__init__()
+                self.first_started = threading.Event()
+                self.release_first = threading.Event()
+
+            def iter_turn_text(self, prompt: str) -> Iterator[str]:
+                self.prompts.append(prompt)
+                if len(self.prompts) == 1:
+                    self.first_started.set()
+                    self.release_first.wait(timeout=2)
+                yield "😊 ok"
+
+        os.environ["DOTTY_KID_MODE"] = "false"
+        client = GatedClient()
+        provider = LLMProvider({}, client=client)  # type: ignore[arg-type]
+        outputs: dict[str, list[str]] = {}
+
+        def run(session, text):
+            outputs[text] = list(provider.response(
+                session, [{"role": "user", "content": text}],
+            ))
+
+        threads = []
+        for i, (session, text) in enumerate(texts_by_session):
+            t = threading.Thread(target=run, args=(session, text))
+            t.start()
+            threads.append(t)
+            if i == 0:
+                self.assertTrue(client.first_started.wait(timeout=1))
+            time.sleep(0.05)  # let it claim its place in the queue
+        client.release_first.set()
+        for t in threads:
+            t.join(timeout=2)
+        return client, outputs
+
+    def test_older_queued_turn_is_skipped(self):
+        client, outputs = self._run(
+            [("s", "running"), ("s", "stale"), ("s", "latest")])
+        self.assertEqual(outputs["stale"], [])
+        self.assertTrue(outputs["latest"])
+        self.assertEqual(len(client.prompts), 2)
+        self.assertNotIn("stale", "".join(client.prompts))
+
+    def test_other_sessions_are_not_affected(self):
+        client, outputs = self._run([("a", "running"), ("a", "queued-a"), ("b", "queued-b")])
+        self.assertTrue(outputs["queued-a"])
+        self.assertTrue(outputs["queued-b"])
+        self.assertEqual(len(client.prompts), 3)
+
+    def test_empty_turn_does_not_supersede_a_real_one(self):
+        client, outputs = self._run([("s", "running"), ("s", "real"), ("s", "   ")])
+        self.assertTrue(outputs["real"])
+        self.assertEqual(len(client.prompts), 2)
+
+    def test_sequential_turns_all_run(self):
+        os.environ["DOTTY_KID_MODE"] = "false"
+        client = FakeClient()
+        client.script_turn(["😊 a"])
+        client.script_turn(["😊 b"])
+        provider = LLMProvider({}, client=client)  # type: ignore[arg-type]
+        self.assertTrue(list(provider.response("s", [{"role": "user", "content": "a"}])))
+        self.assertTrue(list(provider.response("s", [{"role": "user", "content": "b"}])))
+        self.assertEqual(provider._latest_turn, {})
+
+
+class TestPrewarm(unittest.TestCase):
+    """#181: wake the voice model when a device connects."""
+
+    def _provider(self, config=None):
+        return LLMProvider(config or {}, client=FakeClient())  # type: ignore[arg-type]
+
+    def test_posts_a_one_token_completion_to_the_configured_url(self):
+        calls = []
+        done = threading.Event()
+
+        class Resp:
+            def close(self):
+                done.set()
+
+        def fake_urlopen(req, timeout=None):
+            calls.append((req.full_url, json.loads(req.data), timeout))
+            return Resp()
+
+        provider = self._provider({"prewarm_url": "http://llm.test/v1/chat/completions"})
+        with patch("pi_voice.pi_voice.urllib.request.urlopen", fake_urlopen):
+            self.assertTrue(provider.prewarm())
+            self.assertTrue(done.wait(timeout=2))
+        url, body, _ = calls[0]
+        self.assertEqual(url, "http://llm.test/v1/chat/completions")
+        self.assertEqual(body["model"], "qwen3.5:4b")
+        self.assertEqual(body["max_tokens"], 1)
+
+    def test_rate_limited(self):
+        provider = self._provider({"prewarm_url": "http://llm.test/v1/chat/completions"})
+        with patch("pi_voice.pi_voice.urllib.request.urlopen"):
+            self.assertTrue(provider.prewarm())
+            self.assertFalse(provider.prewarm())
+
+    def test_off_without_url_or_when_disabled(self):
+        with patch.dict(os.environ, {"BRIDGE_URL": "", "VISION_BRIDGE_URL": "",
+                                     "DOTTY_PREWARM_URL": ""}):
+            self.assertFalse(self._provider().prewarm())
+        off = self._provider({"prewarm": False, "prewarm_url": "http://x/v1"})
+        self.assertFalse(off.prewarm())
+
+    def test_url_derived_from_sibling_host(self):
+        with patch.dict(os.environ, {"VISION_BRIDGE_URL": "http://10.0.0.5:8090",
+                                     "BRIDGE_URL": "", "DOTTY_PREWARM_URL": ""}):
+            self.assertEqual(
+                self._provider()._prewarm_url,
+                "http://10.0.0.5:8080/v1/chat/completions",
+            )
+
+    def test_failure_never_raises(self):
+        provider = self._provider({"prewarm_url": "http://llm.test/x"})
+        with patch("pi_voice.pi_voice.urllib.request.urlopen", side_effect=OSError("down")):
+            self.assertTrue(provider.prewarm())
+            time.sleep(0.1)
+
+
 class TestNewSessionLifecycle(unittest.TestCase):
     def test_first_turn_skips_new_session(self):
         os.environ["DOTTY_KID_MODE"] = "true"
diff --git a/custom-providers/xiaozhi-patches/websocket_server.py b/custom-providers/xiaozhi-patches/websocket_server.py
index 4d23e03..f15f87b 100644
--- a/custom-providers/xiaozhi-patches/websocket_server.py
+++ b/custom-providers/xiaozhi-patches/websocket_server.py
@@ -163,6 +163,15 @@ async def _handle_connection(self, websocket: websockets.ServerConnection):
                 self.logger.bind(tag=TAG).info(
                     f"restored device state {_restored!r} for {_dotty_dev_id}"
                 )
+        # DOTTY-PATCH: wake the voice model as soon as a device connects, so a
+        # model llama-swap unloaded is loading while the person is still
+        # talking (#181). Never lets a warm-up problem touch the connection.
+        try:
+            _prewarm = getattr(self._llm, "prewarm", None)
+            if callable(_prewarm):
+                _prewarm()
+        except Exception as e:
+            self.logger.bind(tag=TAG).warning(f"LLM prewarm skipped: {e}")
         try:
             await handler.handle_connection(websocket)
         except Exception as e:
diff --git a/tests/test_whisper_noise_close.py b/tests/test_whisper_noise_close.py
new file mode 100644
index 0000000..07d136b
--- /dev/null
+++ b/tests/test_whisper_noise_close.py
@@ -0,0 +1,100 @@
+"""#105: close the session after a run of noise-like ASR turns.
+
+AI-assisted (Claude). Maintainer direction 2026-10-06: close the session on
+room noise (touch to wake) rather than keep listening. Real quiet speech must
+survive: it scored no_speech <= 0.32 and avg_logprob >= -0.9 on the robot.
+"""
+import asyncio
+import types
+
+import numpy as np
+
+from tests.test_whisper_no_speech_gate import _load
+
+
+def _provider(module, config=None):
+    config = config or {}
+    p = module.ASRProvider.__new__(module.ASRProvider)
+    p.no_speech_threshold = module._no_speech_threshold(config)
+    p.hotwords = None
+    p.noise_no_speech = module._float_option(config, "noise_no_speech", module.DEFAULT_NOISE_NO_SPEECH)
+    p.noise_logprob = module._float_option(config, "noise_avg_logprob", module.DEFAULT_NOISE_LOGPROB)
+    p.noise_runs = module.NoiseRunTracker(module._int_option(config, "noise_turn_limit", 3))
+    return p
+
+
+def _run_turns(module, turns, closes):
+    async def go():
+        provider = _provider(module)
+        module._close_session_for_noise = lambda sid, n: closes.append(sid) or asyncio.sleep(0)
+        outs = []
+        for args in turns:
+            segs = [types.SimpleNamespace(text=args[0], no_speech_prob=args[1], avg_logprob=args[2])]
+            provider._transcribe_blocking = lambda audio, segs=segs: (segs, types.SimpleNamespace(language_probability=1.0))
+            art = types.SimpleNamespace(pcm_bytes=np.zeros(16000, dtype=np.int16).tobytes(), file_path="x.wav")
+            out, _ = await provider.speech_to_text([], "s", artifacts=art)
+            outs.append(out)
+        await asyncio.sleep(0)
+        return outs
+    return asyncio.run(go())
+
+
+NOISE = (" and then the senator said", 0.5, -0.4)
+QUIET_CHILD = (" can I have a story", 0.3, -0.5)
+
+
+def test_three_noise_turns_close_the_session_and_blank_the_last():
+    module, closes = _load(), []
+    outs = _run_turns(module, [NOISE, NOISE, NOISE], closes)
+    assert closes == ["s"]
+    assert outs[0]["content"] and outs[1]["content"]
+    assert outs[2] == {"content": ""}
+
+
+def test_quiet_real_speech_never_counts():
+    module, closes = _load(), []
+    outs = _run_turns(module, [QUIET_CHILD] * 6, closes)
+    assert closes == []
+    assert all(o["content"] for o in outs)
+
+
+def test_real_turn_between_noise_resets_the_run():
+    module, closes = _load(), []
+    _run_turns(module, [NOISE, NOISE, QUIET_CHILD, NOISE, NOISE], closes)
+    assert closes == []
+
+
+def test_rejected_and_empty_turns_count_as_noise():
+    module, closes = _load(), []
+    # no_speech 0.81 is dropped by the existing gate -> empty content
+    _run_turns(module, [(" Thank you.", 0.81, -0.3)] * 3, closes)
+    assert closes == ["s"]
+
+
+def test_low_confidence_decode_counts():
+    module = _load()
+    assert module._is_noise_like("mumble", 0.1, -1.4, 0.45, -1.0)
+    assert not module._is_noise_like("hello dotty", 0.1, -0.9, 0.45, -1.0)
+
+
+def test_runs_far_apart_are_separate_episodes():
+    module = _load()
+    tracker = module.NoiseRunTracker(3, window_s=90)
+    assert not tracker.record("s", True, 0)
+    assert not tracker.record("s", True, 10)
+    assert not tracker.record("s", True, 200)  # restarts: count 1
+    assert not tracker.record("s", True, 210)
+    assert tracker.record("s", True, 220)
+
+
+def test_limit_zero_disables():
+    module = _load()
+    tracker = module.NoiseRunTracker(0)
+    assert not any(tracker.record("s", True, i) for i in range(10))
+
+
+def test_sessions_are_tracked_separately():
+    module = _load()
+    tracker = module.NoiseRunTracker(3)
+    for sid in ("a", "b", "a", "b"):
+        assert not tracker.record(sid, True, 0)