diff --git a/CONNECTION_STABILITY_REVIEW.md b/CONNECTION_STABILITY_REVIEW.md
new file mode 100644
index 00000000..7c0f4866
--- /dev/null
+++ b/CONNECTION_STABILITY_REVIEW.md
@@ -0,0 +1,283 @@
+# PAI connection-stability evaluation
+
+**Status:** all 13 findings below are fixed on this branch. The analysis is kept
+as the rationale for each change; see `git log` for the commits. Verified with
+the full suite (1721 tests) plus 21 new regression tests, each confirmed to fail
+against the original code.
+
+**Scope:** defects in the PAI codebase that could cause or worsen repeated loss of
+connection to a Paradox panel, and slow/failed recovery afterwards.
+**Base:** `ParadoxAlarmInterface/pai` @ `be1e46e` (branch `dev`).
+**Method:** manual read of the connect/poll/reconnect paths — `main.py`, `paradox.py`,
+`connections/*`, `lib/handlers.py`, `lib/async_message_manager.py`, `lib/stun.py`.
+
+Findings are ordered by how likely they are to be contributing to a "regularly loses
+connection" symptom. Line numbers are against the base commit.
+
+---
+
+## Tier 1 — most likely to be driving the symptom
+
+### 1. `IO_TIMEOUT` from `pai.conf` is silently ignored on every request path
+
+`cfg.IO_TIMEOUT` is used as a **default argument value**, which Python evaluates once at
+import time:
+
+- `paradox/paradox.py:497` — `send_wait(..., timeout=cfg.IO_TIMEOUT, ...)`
+- `paradox/lib/async_message_manager.py:42` — `wait_for_message(..., timeout=cfg.IO_TIMEOUT)`
+- `paradox/lib/handlers.py:97` — `wait_until_complete(self, handler, timeout=cfg.IO_TIMEOUT)`
+- `paradox/connections/ip/connection.py:98` — `wait_for_ip_message(self, timeout=cfg.IO_TIMEOUT)`
+
+`paradox/console_scripts/pai_run.py` imports `paradox.main` (which imports `Paradox`)
+**before** `main()` calls `cfg.load()`. So all four defaults freeze at the built-in
+default of `0.5` (`paradox/config.py:93`), regardless of what the user configures.
+
+Consequence: the single most useful tuning knob for a marginal link is a no-op. Anyone
+raising `IO_TIMEOUT` to cope with a slow IP150 or a busy panel sees no change and
+concludes the problem is elsewhere.
+
+Note the call sites that read `cfg.IO_TIMEOUT` at *call* time — `protocol_base.py:56`,
+`hardware/prt3/panel.py` — do honour the config, so behaviour is inconsistent between
+subsystems.
+
+Fix: sentinel default (`timeout=None`) resolved to `cfg.IO_TIMEOUT` inside the function.
+
+### 2. The poll cycle has no upper bound, and degrades into a poll storm
+
+`Paradox.loop()` (`paradox/paradox.py:408-441`):
+
+```python
+tstart = time.time()
+result = await asyncio.gather(*self.panel.get_status_requests())
+...
+max_wait_time = max((tstart + cfg.KEEP_ALIVE_INTERVAL) - time.time(), 0)
+await asyncio.wait_for(self.loop_wait_event.wait(), max_wait_time)
+```
+
+Every status request is serialised behind `self.request_lock` (`paradox.py:508`) and
+`send_wait` retries **5** times with a per-attempt budget of `timeout * 2`.
+
+For EVO, `status_request_addresses` is 14 addresses (`hardware/evo/parsers.py:180`,
+keys `1-11, 16, 57, 58`); Spectra/Magellan is 7. Worst case for EVO:
+
+    14 addresses x 5 retries x (2 x 0.5 s) = ~70 s per cycle
+
+against a `KEEP_ALIVE_INTERVAL` of 10 s. Once a cycle overruns the interval,
+`max_wait_time` clamps to `0` and the loop immediately starts another full poll with no
+idle gap — so a panel that is *momentarily* slow gets hit with back-to-back polls, which
+is exactly the wrong response. There is no "skip this cycle, we're behind" guard and no
+cap on cycle duration.
+
+Fix: bound the whole cycle (e.g. `asyncio.wait_for(gather(...), KEEP_ALIVE_INTERVAL)`),
+lower `retries` for status polls, and enforce a minimum idle gap between cycles.
+
+### 3. Reconnect backoff is `2 ^ retry` — bitwise XOR, not exponentiation
+
+`paradox/main.py:124`:
+
+```python
+retry_time_wait = 2 ^ retry
+retry_time_wait = 30 if retry_time_wait > 30 else retry_time_wait
+```
+
+`^` is XOR in Python. The actual wait sequence is:
+
+| retry | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 | 10 | 11 | 12 |
+|---|---|---|---|---|---|---|---|---|---|---|---|---|
+| wait (s) | 3 | **0** | 1 | 6 | 7 | 4 | 5 | 10 | 11 | 8 | 9 | 14 |
+
+It is not monotonic, it never approaches the 30 s cap in any realistic number of
+attempts, and the **second** retry waits **zero seconds**. An IP150 accepts a single
+session at a time and needs time to tear the old one down; reconnecting immediately is
+the reliable way to be refused and stay refused. Compounds directly with finding 4.
+
+Fix: `retry_time_wait = min(2 ** retry, 30)`, ideally with jitter.
+
+### 4. Failed IP connection attempts leak the previous socket
+
+`MultiAttemptConnection.connect()` (`paradox/connections/ip/connection.py:24-49`) retries
+`_try_connect()` up to 3 times. But `_try_connect` for both local and STUN connections
+(`ip/connection.py:113-121` and `143-151`) assigns `self._protocol` as soon as
+`create_connection` returns, *then* runs the module handshake:
+
+```python
+_, self._protocol = await asyncio.get_running_loop().create_connection(...)
+await IPModuleConnectCommand(self).execute()   # <- can raise
+self.connected = True
+```
+
+If the handshake raises (timeout, auth failure), the exception is caught in `connect()`
+and the loop retries — overwriting `self._protocol` with a new connection **without
+closing the previous transport**. The orphaned socket is only closed whenever GC gets
+around to it; `ConnectionProtocol.__del__` (`protocol_base.py:89`) deliberately does not
+close the transport.
+
+Result: up to 3 orphaned TCP sessions per `connect()` call, each still occupying the IP
+module's single session slot — so the retries are competing with PAI's own leaked
+sockets. There is also **no delay** between the three attempts.
+
+Fix: `await self.close()` in the exception path before retrying, and sleep between tries.
+
+---
+
+## Tier 2 — STUN / paradoxmyhome path (blocking I/O on the event loop)
+
+### 5. `refresh_session_if_required()` does blocking socket I/O on the event loop
+
+`StunSession.refresh_session_if_required()` (`connections/ip/stun_session.py:141-153`) is
+called **synchronously** from `StunIPConnection.write()` (`ip/connection.py:136-141`),
+which runs on the asyncio event loop. It calls `stun_control.send_refresh_request()` and
+`receive_response()`.
+
+`StunClient` (`paradox/lib/stun.py:305-311`) creates a plain blocking socket and never
+calls `settimeout()`; `receive_response()` (`lib/stun.py:339-340`) is a bare
+`self.sock.recv(2048)`.
+
+So every ~500 s, a STUN user's entire event loop can block indefinitely inside `write()`
+on a half-dead TURN control socket. Nothing else runs: no `data_received`, no keepalive,
+no MQTT. When it finally returns, every pending request has timed out at once and the
+session is torn down. This is a strong match for periodic, roughly-regular dropouts on
+paradoxmyhome setups.
+
+`StunClient.__init__` also does a blocking `sock.connect()` (`lib/stun.py:311`), reached
+from `async def _stun_tcp_change_request` (`stun_session.py:110`).
+
+Fix: `sock.settimeout(...)` on all STUN sockets, and run the refresh in an executor (or
+convert to non-blocking) rather than inline in `write()`.
+
+### 6. `time.sleep(5)` inside an async function, and an unbounded HTTP call
+
+`StunSession._get_site_info` (`stun_session.py:180-198`) is `async def` but calls
+`time.sleep(5)` (line 196) on retry — blocking the event loop for up to 25 s across 5
+attempts. The `requests.get` on line 187 has **no timeout**, so a stalled SWAN API call
+hangs the connect path forever with no recovery.
+
+Fix: `await asyncio.sleep(5)`; pass `timeout=` to `requests.get`.
+
+### 7. STUN response parsing assumes one `recv()` returns a whole message
+
+`lib/stun.py:339-345`:
+
+```python
+buf = self.sock.recv(2048)
+...
+assert len(attributes) == body_length
+```
+
+TCP does not guarantee message boundaries. A segmented STUN response trips the assert,
+which is swallowed by the generic `except Exception` in `MultiAttemptConnection.connect`
+and reported as an unexplained connect failure. Needs a read-until-complete loop.
+
+### 8. Failed STUN refresh does not mark the connection dead
+
+`stun_session.py:151` sets `self.connected = False` — but `StunSession` has no
+`connected` attribute (see `__init__`, lines 38-48) and nothing reads it. The assignment
+just creates a stray attribute. The intent (propagate the dead session) is lost; only the
+raised `StunSessionRefreshFailed` carries the signal.
+
+---
+
+## Tier 3 — smaller, still real
+
+### 9. `busy.release()` can be called without holding the lock
+
+`paradox/paradox.py:412` / `:428`:
+
+```python
+try:
+    await self.busy.acquire()      # inside the try
+    ...
+finally:
+    self.busy.release()            # runs even if acquire() never succeeded
+```
+
+`busy` is contended — the IP interface holds it at
+`interfaces/ip_interface/client_connection.py:173`. If `acquire()` is cancelled (shutdown,
+or a BabyWare client holding the lock while PAI is stopping), the `finally` raises
+`RuntimeError: Lock is not acquired`, which escapes `loop()` and is caught by the generic
+handler in `main.py:158` — turning a clean stop into an exception-driven restart.
+
+Fix: `async with self.busy:` around the body.
+
+### 10. `asyncio.gather` abandons sibling requests on first failure
+
+`paradox.py:413`. `gather` without `return_exceptions` propagates the first
+`StatusRequestException` but does **not** cancel the remaining status requests. They stay
+queued on `request_lock` and overlap with the next cycle's requests, so a single slow
+address compounds into cross-cycle contention rather than being isolated.
+
+### 11. A failed close leaves the connection in a half-torn-down state
+
+`connections/connection.py:52-56`:
+
+```python
+async def close(self):
+    if self._protocol:
+        await self._protocol.close()   # can raise
+        self._protocol = None
+    self.connected = False
+```
+
+`ConnectionProtocol.close()` (`protocol_base.py:47-57`) awaits `self._closed`, and
+`connection_lost` sets an **exception** on that future when the transport died with an
+error (`protocol_base.py:80-83`). Awaiting it therefore re-raises — which is the normal
+case when closing after a fault. Neither `_protocol = None` nor `connected = False` runs.
+The `asyncio.wait_for(..., cfg.IO_TIMEOUT)` on the same line can also raise
+`TimeoutError` with the same effect.
+
+Fix: wrap in `try/finally` so the state reset always happens.
+
+### 12. A status parse failure crashes the merge instead of counting as a missing reply
+
+`Panel.handle_status` (`hardware/panel.py:334-355`) returns `None` when the address has no
+parser or the parse throws. `request_status` returns that `None` straight through
+(`evo/panel.py:233`, `spectra_magellan/panel.py:239`), and `deep_merge`
+(`lib/utils.py:66-69`) then calls `d2.items()` on `None` → `AttributeError`.
+
+That is caught by the catch-all `except Exception: logger.exception("Loop")`
+(`paradox.py:425`), so `replies_missing` is **not** incremented — the cycle's entire
+status update is dropped silently and the health counter never notices.
+
+### 13. `disconnect()` can construct a connection object during shutdown
+
+`Paradox.disconnect()` (`paradox.py:913-920`) reaches `self.connection`, which is a lazy
+property that *builds* a `Connection` and registers handlers when `_connection is None`
+(`paradox.py:122-174`). `main._run`'s `exit_handler` calls `disconnect()`
+unconditionally, so shutting down before a first connection constructs one just to ask
+whether it is connected. Harmless today, but it makes the shutdown path depend on config
+validity (`raise AssertionError(f"Invalid connection type: ...")` at `paradox.py:170`).
+
+---
+
+## Behaviour changes worth knowing about
+
+- **Reconnect backoff is now 2, 4, 8, 16, 30, 30 s** (was 3, 0, 1, 6, 7, ...).
+  Recovery from a brief blip is marginally slower; recovery from a real outage
+  is much more reliable, because PAI stops hammering the module.
+- **IP connect attempts are now 5 s apart** (`CONNECT_RETRY_DELAY`), so a failing
+  `connect()` takes ~10 s longer before giving up and handing back to the main
+  retry loop.
+- **Raising `IO_TIMEOUT` now actually takes effect.** Anyone who had raised it to
+  work around finding 1 should re-check the value: it was being ignored, so the
+  configured number has never been exercised.
+- **A failed IP attempt now closes its STUN session**, so the next attempt
+  re-fetches the SWAN site info instead of reusing a possibly stale `xoraddr`.
+  That costs one extra HTTPS round trip per retry.
+- **The poll cycle is bounded** by `Panel.status_cycle_budget`. A cycle that
+  exceeds it is cancelled and counted as a missing reply, so a wedged cycle now
+  surfaces as "Replies missing" and a reconnect instead of silence. PRT3
+  overrides the budget because its single virtual address expands into one
+  request per area and zone.
+
+## Diagnostics worth collecting first
+
+To confirm which path is in play, set `LOGGING_LEVEL_CONSOLE = 10` (DEBUG) and look for:
+
+- `send/receive timeout in ...` lines from `paradox.py:551` — frequency tells you whether
+  finding 1/2 is biting.
+- `Loop: Replies missing: N` — how close each cycle gets to the forced-disconnect
+  threshold of 3.
+- `Connection recovered after ... down` / `Panel connection ended after ... up`
+  (`main.py:105`, `:114`) — gives the actual uptime/outage distribution.
+- `Connecting. Try n/3` bursts with no gap — finding 4.
+- `STUN Session Refresh` immediately preceding a stall — finding 5.
diff --git a/paradox/connections/connection.py b/paradox/connections/connection.py
index 8bdc232b..f8add8f3 100644
--- a/paradox/connections/connection.py
+++ b/paradox/connections/connection.py
@@ -49,10 +49,18 @@ def write(self, data: bytes):
             raise ConnectionError("Not connected")
 
     async def close(self):
-        if self._protocol:
-            await self._protocol.close()
+        protocol = self._protocol
+        try:
+            if protocol is not None:
+                await protocol.close()
+        finally:
+            # ConnectionProtocol.close() re-raises whatever killed the
+            # transport -- the normal case when closing after a fault -- and
+            # can also time out waiting for it to settle. Either way the
+            # protocol is spent, so the reset has to happen regardless or the
+            # next connect attempt inherits a dead one.
             self._protocol = None
-        self.connected = False
+            self.connected = False
 
     def variable_message_length(self, mode):
         if self._protocol is not None:
diff --git a/paradox/connections/ip/connection.py b/paradox/connections/ip/connection.py
index 6eb8a7d8..e75480b8 100644
--- a/paradox/connections/ip/connection.py
+++ b/paradox/connections/ip/connection.py
@@ -1,10 +1,10 @@
 from abc import ABC, abstractmethod
 import asyncio
 import logging
+from typing import Optional
 
 from construct import Container
 
-from paradox.config import config as cfg
 from paradox.connections.connection import Connection
 from paradox.connections.handler import IPConnectionHandler
 from paradox.connections.ip.commands import IPModuleConnectCommand
@@ -17,6 +17,12 @@
 logger = logging.getLogger("PAI").getChild(__name__)
 
 
+#: Pause between connection attempts. The IP module accepts a single session
+#: at a time and only frees the slot once it notices the old socket is gone,
+#: so retrying immediately is a reliable way to be refused three times.
+CONNECT_RETRY_DELAY = 5.0
+
+
 class MultiAttemptConnection(Connection):
     async def connect(self) -> bool:
         tries = 1
@@ -25,9 +31,15 @@ async def connect(self) -> bool:
 
         while tries <= max_tries:
             logger.info("Connecting. Try %d/%d" % (tries, max_tries))
+            succeeded = False
             try:
-                await self._try_connect()
-                return True
+                try:
+                    await self._try_connect()
+                    succeeded = True
+                    return True
+                finally:
+                    if not succeeded:
+                        await self._discard_failed_attempt()
             except asyncio.TimeoutError as e:
                 logger.error(
                     "Timeout while establishing connection (try %d/%d): %s"
@@ -46,9 +58,29 @@ async def connect(self) -> bool:
                 )
 
             tries += 1
+            if tries <= max_tries:
+                await asyncio.sleep(CONNECT_RETRY_DELAY)
 
         return False
 
+    async def _discard_failed_attempt(self) -> None:
+        """Tear down a half-open attempt before retrying.
+
+        ``_try_connect`` stores the protocol as soon as the socket opens and
+        only then runs the module handshake, so a handshake failure leaves an
+        open socket owned by nothing. Overwriting ``_protocol`` on the next try
+        would strand it -- ``ConnectionProtocol`` deliberately does not close
+        the transport on ``__del__`` -- leaving PAI to compete with its own
+        orphans for the module's single session slot.
+        """
+        try:
+            await self.close()
+        except Exception:
+            logger.debug(
+                "Ignoring error while discarding a failed connection attempt",
+                exc_info=True,
+            )
+
     @abstractmethod
     async def _try_connect(self):
         raise NotImplementedError("Implement in a subclass")
@@ -95,7 +127,9 @@ def on_ip_message(self, container: Container):
             self.ip_handler_registry.handle(container)
         )
 
-    async def wait_for_ip_message(self, timeout=cfg.IO_TIMEOUT) -> Container:
+    async def wait_for_ip_message(self, timeout=None) -> Container:
+        # ``None`` defers to cfg.IO_TIMEOUT inside wait_until_complete, which
+        # reads it at call time rather than at import time.
         future = FutureHandler()
         return await self.ip_handler_registry.wait_until_complete(future, timeout)
 
@@ -132,6 +166,7 @@ def __init__(self, site_id, email, panel_serial, password):
         super().__init__(password)
 
         self.stun_session = StunSession(site_id, email, panel_serial)
+        self._refresh_task: Optional[asyncio.Task] = None
 
     def on_connection_loss(self):
         super().on_connection_loss()
@@ -139,14 +174,56 @@ def on_connection_loss(self):
         if self.stun_session is not None:
             self.stun_session.close()
 
+    async def close(self):
+        # The tunnel socket is the transport's, but the control socket is
+        # ours. A connect attempt that opens the STUN session and then fails
+        # would otherwise leak it, and the TURN allocation with it.
+        try:
+            await super().close()
+        finally:
+            if self.stun_session is not None:
+                self.stun_session.close()
+
     def write(self, data: bytes):
         """Write data to socket"""
 
-        if self.stun_session is not None:
-            self.stun_session.refresh_session_if_required()
+        self._schedule_session_refresh()
 
         return super().write(data)
 
+    def _schedule_session_refresh(self) -> None:
+        """Renew the TURN allocation in the background, if it is due.
+
+        The refresh used to run inline here, doing blocking socket I/O on the
+        event loop: every ~500 s PAI stopped servicing the panel, MQTT and
+        every other interface until the TURN server answered. The allocation
+        still has ~100 s of life left when a refresh becomes due, so letting
+        this write proceed while the renewal runs in a thread is both safe and
+        what keeps a slow TURN server from looking like a panel dropout.
+        """
+        if self.stun_session is None or not self.stun_session.refresh_required():
+            return
+
+        if self._refresh_task is not None and not self._refresh_task.done():
+            return  # One already in flight; do not queue a second.
+
+        self._refresh_task = asyncio.get_running_loop().create_task(
+            self._refresh_session()
+        )
+
+    async def _refresh_session(self) -> None:
+        loop = asyncio.get_running_loop()
+        try:
+            await loop.run_in_executor(None, self.stun_session.refresh_session)
+        except Exception:
+            # The allocation is gone or the control socket is dead. Drop the
+            # link so the main loop reconnects, rather than writing into a
+            # tunnel that has already stopped forwarding.
+            logger.error(
+                "STUN session refresh failed, dropping connection", exc_info=True
+            )
+            await self.close()
+
     async def _try_connect(self) -> None:
         await self.stun_session.connect()
         _, self._protocol = await asyncio.get_running_loop().create_connection(
diff --git a/paradox/connections/ip/stun_session.py b/paradox/connections/ip/stun_session.py
index 4c6a00ed..ef5525e6 100644
--- a/paradox/connections/ip/stun_session.py
+++ b/paradox/connections/ip/stun_session.py
@@ -12,6 +12,13 @@
 
 logger = logging.getLogger("PAI").getChild(__name__)
 
+#: TURN allocations are handed out with a 600 s lifetime; renew with headroom.
+SESSION_REFRESH_INTERVAL = 500
+
+#: Bound on the SWAN site lookup. Without one a stalled HTTP call hangs the
+#: connect path forever.
+SITE_INFO_HTTP_TIMEOUT = 20
+
 SENSITIVE_SITE_INFO_KEYS = {
     "panelSerial": mask_secret,
     "email": mask_email,
@@ -138,22 +145,35 @@ def _select_module(self):
     def get_socket(self):
         return self.stun_tunnel.sock
 
-    def refresh_session_if_required(self) -> None:
+    def refresh_required(self) -> bool:
+        """Whether the TURN allocation is close enough to expiry to renew."""
         if self.site_info is None or self.connection_timestamp == 0:
-            return
-
-        # Refresh session if required
-        if time.time() - self.connection_timestamp >= 500:
-            logger.info("STUN Session Refresh")
-            self.stun_control.send_refresh_request()
-            stun_r = self.stun_control.receive_response()
-            if stun.is_error(stun_r):
-                self.connected = False
-                raise StunSessionRefreshFailed(
-                    f"STUN Session Refresh failed: {stun.get_error(stun_r)}"
-                )
+            return False
 
-            self.connection_timestamp = time.time()
+        return time.time() - self.connection_timestamp >= SESSION_REFRESH_INTERVAL
+
+    def refresh_session(self) -> None:
+        """Renew the TURN allocation.
+
+        Blocking: this talks to the control socket synchronously. Call it from
+        an executor, never from the event loop -- it used to run inline in
+        ``write()``, where a slow or half-dead TURN server stalled every
+        interface PAI was running until the socket gave up.
+        """
+        logger.info("STUN Session Refresh")
+        self.stun_control.send_refresh_request()
+        stun_r = self.stun_control.receive_response()
+        if stun.is_error(stun_r):
+            raise StunSessionRefreshFailed(
+                f"STUN Session Refresh failed: {stun.get_error(stun_r)}"
+            )
+
+        self.connection_timestamp = time.time()
+
+    def refresh_session_if_required(self) -> None:
+        """Blocking refresh-if-due. Retained for callers outside the event loop."""
+        if self.refresh_required():
+            self.refresh_session()
 
     def close(self):
         self.site_info = None
@@ -185,7 +205,10 @@ async def _get_site_info(email, siteid):
             req = await loop.run_in_executor(
                 None,
                 lambda: requests.get(
-                    URL, headers=headers, params={"email": email, "name": siteid}
+                    URL,
+                    headers=headers,
+                    params={"email": email, "name": siteid},
+                    timeout=SITE_INFO_HTTP_TIMEOUT,
                 ),
             )
             if req.status_code == 200:
@@ -193,7 +216,9 @@ async def _get_site_info(email, siteid):
 
             logger.warning("Unable to get site info. Retrying...")
             tries -= 1
-            time.sleep(5)
+            # await, not time.sleep: this is a coroutine, and blocking here
+            # stalled the whole event loop for up to 25 s.
+            await asyncio.sleep(5)
 
         return None
 
diff --git a/paradox/hardware/panel.py b/paradox/hardware/panel.py
index 622321a0..2cedd835 100644
--- a/paradox/hardware/panel.py
+++ b/paradox/hardware/panel.py
@@ -328,6 +328,26 @@ def initialize_communication(self, password):
     def get_status_requests(self) -> typing.Iterable[typing.Awaitable]:
         return (self.request_status(i) for i in self.status_request_addresses)
 
+    @property
+    def status_cycle_budget(self) -> float:
+        """Wall-clock bound for one full status poll, in seconds.
+
+        The poll loop abandons a cycle that exceeds this, so it has to allow
+        for what a healthy-but-slow panel legitimately needs: the requests run
+        one at a time behind ``request_lock``, each waiting up to
+        ``IO_TIMEOUT * 2`` for its reply, with room for one retry. It is a
+        backstop against a wedged cycle, not a latency target -- which is why
+        it is generous rather than tight.
+
+        Panels whose virtual addresses expand into several requests each must
+        override this; see PRT3Panel.
+        """
+        per_request = cfg.IO_TIMEOUT * 2 * 2  # reply window, plus one retry
+        return max(
+            cfg.KEEP_ALIVE_INTERVAL,
+            per_request * len(list(self.status_request_addresses)),
+        )
+
     @abstractmethod
     async def request_status(self, nr) -> typing.Optional[Container]:
         raise NotImplementedError("override request_status in a subclass")
diff --git a/paradox/hardware/prt3/panel.py b/paradox/hardware/prt3/panel.py
index 6f3469b4..e892f50e 100644
--- a/paradox/hardware/prt3/panel.py
+++ b/paradox/hardware/prt3/panel.py
@@ -93,6 +93,17 @@ class PRT3Panel(Panel):
     # all configured areas and zones internally.
     status_request_addresses = [0]
 
+    @property
+    def status_cycle_budget(self) -> float:
+        """One request per area and per zone, not one per status address.
+
+        The base implementation would size the budget from
+        ``status_request_addresses``, which is a single virtual address here,
+        and cut every cycle short on a panel with many zones.
+        """
+        elements = cfg.PRT3_MAX_AREAS + cfg.PRT3_MAX_ZONES
+        return max(cfg.KEEP_ALIVE_INTERVAL, cfg.IO_TIMEOUT * elements * 2)
+
     def __init__(self, core):
         # variable_message_length=False: PRT3Protocol.variable_message_length()
         # is a no-op; the Panel base class must not try to manage lengths.
diff --git a/paradox/lib/async_message_manager.py b/paradox/lib/async_message_manager.py
index 73f8ba6d..9d63a6f0 100644
--- a/paradox/lib/async_message_manager.py
+++ b/paradox/lib/async_message_manager.py
@@ -4,7 +4,6 @@
 
 from construct import Container
 
-from paradox.config import config as cfg
 from paradox.lib.handlers import FutureHandler, HandlerRegistry, PersistentHandler
 
 logger = logging.getLogger("PAI").getChild(__name__)
@@ -39,8 +38,10 @@ def __init__(self):
     async def wait_for_message(
         self,
         check_fn: Optional[Callable[[Container], bool]] = None,
-        timeout=cfg.IO_TIMEOUT,
+        timeout=None,
     ) -> Container:
+        # ``None`` defers to cfg.IO_TIMEOUT inside wait_until_complete, which
+        # reads it at call time. See the note there.
         return await self.handler_registry.wait_until_complete(
             FutureHandler(check_fn), timeout
         )
diff --git a/paradox/lib/handlers.py b/paradox/lib/handlers.py
index 73070e1d..96efd26c 100644
--- a/paradox/lib/handlers.py
+++ b/paradox/lib/handlers.py
@@ -94,7 +94,14 @@ def remove_by_name(self, name: str):
         for handler in to_remove:
             self.remove(handler)
 
-    async def wait_until_complete(self, handler: Handler, timeout=cfg.IO_TIMEOUT):
+    async def wait_until_complete(self, handler: Handler, timeout=None):
+        # Resolved per call, never as a default argument value: PAI imports
+        # this module before ``main()`` runs ``cfg.load()``, so a default
+        # argument would freeze the built-in IO_TIMEOUT and silently ignore
+        # whatever the user configured.
+        if timeout is None:
+            timeout = cfg.IO_TIMEOUT
+
         self.append(handler)
         try:
             return await asyncio.wait_for(handler, timeout=timeout)
diff --git a/paradox/lib/stun.py b/paradox/lib/stun.py
index e386a751..b6acbefd 100644
--- a/paradox/lib/stun.py
+++ b/paradox/lib/stun.py
@@ -9,6 +9,14 @@
 
 STUN_PORT = 3478
 
+#: Every STUN message starts with a fixed 20 byte header.
+STUN_HEADER_LENGTH = 20
+
+#: Bound on every blocking STUN socket operation. Without one, a TURN server
+#: that stops answering hangs the caller indefinitely -- and the session
+#: refresh used to run on the event loop, so that hung all of PAI.
+STUN_SOCKET_TIMEOUT = 10.0
+
 FAMILY_IPv4 = b"\x01"
 FAMILY_IPv6 = b"\x02"
 
@@ -306,6 +314,9 @@ def __init__(self, host, port=STUN_PORT):
 
         self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
         self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        # asyncio clears this when the tunnel socket is handed to
+        # create_connection(sock=...), so it only bounds the blocking phase.
+        self.sock.settimeout(STUN_SOCKET_TIMEOUT)
         self.sock.bind(("0.0.0.0", 0))
         self.host = host
         self.port = port
@@ -337,15 +348,33 @@ def send_connection_bind_request(self, connection_id):
         self.sock.send(self.req)
 
     def receive_response(self):
-        buf = self.sock.recv(2048)
-        validate_response(buf, self.transaction_id)
+        header = self._recv_exactly(STUN_HEADER_LENGTH)
+        validate_response(header, self.transaction_id)
 
-        body_length = int(binascii.b2a_hex(buf[2:4]), 16)
-        attributes = buf[20:]
-        assert len(attributes) == body_length
+        body_length = int(binascii.b2a_hex(header[2:4]), 16)
+        attributes = self._recv_exactly(body_length)
 
         return read_attributes(attributes, body_length)
 
+    def _recv_exactly(self, length):
+        """Read exactly ``length`` bytes from the control socket.
+
+        TCP does not preserve message boundaries, so one recv() can return a
+        short read. The previous code asserted the whole response arrived in a
+        single recv(), which turned ordinary segmentation into an unexplained
+        connection failure.
+        """
+        chunks = []
+        remaining = length
+        while remaining > 0:
+            chunk = self.sock.recv(remaining)
+            if not chunk:
+                raise Exception("Connection closed while reading STUN response")
+            chunks.append(chunk)
+            remaining -= len(chunk)
+
+        return b"".join(chunks)
+
     def send_refresh_request(self):
         self.req = build_connection_refresh_request(self.transaction_id)
         self.sock.send(self.req)
diff --git a/paradox/main.py b/paradox/main.py
index 25ee0b9d..3086a253 100755
--- a/paradox/main.py
+++ b/paradox/main.py
@@ -121,8 +121,11 @@ def mark_disconnected():
 
     while alarm is not None:
         logger.info("Starting...")
-        retry_time_wait = 2 ^ retry
-        retry_time_wait = 30 if retry_time_wait > 30 else retry_time_wait
+        # ``^`` is XOR, not exponentiation. The original expression backed off
+        # 3, 0, 1, 6, 7, 4, 5 ... seconds, so the second attempt reconnected
+        # instantly and the sequence never grew. The IP module serves one
+        # session at a time and needs a moment to release the previous one.
+        retry_time_wait = min(2**retry, 30)
 
         try:
             if await alarm.full_connect():
diff --git a/paradox/paradox.py b/paradox/paradox.py
index 871366bb..84c5d061 100644
--- a/paradox/paradox.py
+++ b/paradox/paradox.py
@@ -31,6 +31,10 @@
 
 logger = logging.getLogger("PAI").getChild(__name__)
 
+#: Smallest gap between two status poll cycles. Only reached when a cycle
+#: overruns KEEP_ALIVE_INTERVAL; an explicit refresh still bypasses it.
+MIN_LOOP_IDLE_TIME = 1.0
+
 
 class Paradox:
     def __init__(self, retries=3):
@@ -409,9 +413,14 @@ async def loop(self):
             tstart = time.time()
             if self.run_state == RunState.RUN:
                 try:
-                    await self.busy.acquire()
-                    result = await asyncio.gather(*self.panel.get_status_requests())
-                    merged = deep_merge(*result, extend_lists=True, initializer={})
+                    # ``async with``, not acquire()/release() in a finally: a
+                    # cancelled acquire() left the finally releasing a lock it
+                    # never held, and the RuntimeError turned a clean shutdown
+                    # into an exception-driven restart.
+                    async with self.busy:
+                        merged = await asyncio.wait_for(
+                            self._poll_status(), self._status_cycle_budget()
+                        )
                     asyncio.get_running_loop().call_soon(self._process_status, merged)
                     replies_missing = max(0, replies_missing - 1)
                 except ConnectionError:
@@ -424,15 +433,19 @@ async def loop(self):
                         return
                 except Exception:
                     logger.exception("Loop")
-                finally:
-                    self.busy.release()
 
                 if replies_missing > 0:
                     logger.debug(f"Loop: Replies missing: {replies_missing}")
 
             # cfg.Listen for events
 
-            max_wait_time = max((tstart + cfg.KEEP_ALIVE_INTERVAL) - time.time(), 0)
+            # Floor the idle gap. A cycle that overran KEEP_ALIVE_INTERVAL used
+            # to leave zero wait here, so PAI re-polled a struggling panel back
+            # to back. request_status_refresh() still gets its immediate cycle:
+            # a set event short-circuits the wait whatever the timeout is.
+            max_wait_time = max(
+                (tstart + cfg.KEEP_ALIVE_INTERVAL) - time.time(), MIN_LOOP_IDLE_TIME
+            )
             try:
                 await asyncio.wait_for(self.loop_wait_event.wait(), max_wait_time)
             except asyncio.TimeoutError:
@@ -441,6 +454,50 @@ async def loop(self):
             finally:
                 self.loop_wait_event.clear()
 
+    def _status_cycle_budget(self) -> float:
+        """How long one status poll may run before it is abandoned.
+
+        Status requests are serialised behind ``request_lock`` and ``send_wait``
+        retries five times, so an unbounded EVO poll (14 RAM addresses) can run
+        for minutes against a ``KEEP_ALIVE_INTERVAL`` of 10 s -- with no
+        missing reply counted and no status published the whole time. The panel
+        owns the number, because only it knows how many requests an address
+        expands into.
+        """
+        return self.panel.status_cycle_budget
+
+    async def _poll_status(self) -> dict:
+        """Run one status poll cycle, as an all-or-nothing batch.
+
+        ``asyncio.gather`` propagates the first exception but leaves its
+        siblings running. Those stragglers stay queued on ``request_lock`` and
+        collide with the next cycle's requests, so one slow address compounds
+        into cross-cycle contention instead of staying in its own cycle.
+        """
+        tasks = [
+            asyncio.ensure_future(request)
+            for request in self.panel.get_status_requests()
+        ]
+        try:
+            results = await asyncio.gather(*tasks)
+        except BaseException:
+            for task in tasks:
+                task.cancel()
+            raise
+
+        # handle_status() returns None for a block it cannot parse. Passing
+        # that to deep_merge raises, which the caller logs as a generic loop
+        # error and drops the whole cycle -- including the blocks that did
+        # parse -- without counting a missing reply.
+        parsed = [result for result in results if result is not None]
+        if len(parsed) != len(results):
+            logger.warning(
+                "Discarded %d unparsable status block(s) this cycle",
+                len(results) - len(parsed),
+            )
+
+        return deep_merge(*parsed, extend_lists=True, initializer={})
+
     @staticmethod
     def _process_status(raw_status: Container) -> None:
         status = convert_raw_status(raw_status)
@@ -494,9 +551,15 @@ async def send_wait(
         args=None,
         message=None,
         retries=5,
-        timeout=cfg.IO_TIMEOUT,
+        timeout=None,
         reply_expected=None,
     ) -> Optional[Container]:
+        # Read at call time, not as a default argument value: this module is
+        # imported before main() runs cfg.load(), so a default would freeze the
+        # built-in IO_TIMEOUT and ignore the configured one.
+        if timeout is None:
+            timeout = cfg.IO_TIMEOUT
+
         # Connection closed
         if not self.connection.connected:
             raise ConnectionError("Not connected")
@@ -914,9 +977,16 @@ async def disconnect(self):
         logger.info("Disconnecting from the Alarm Panel")
         self.run_state = RunState.STOP
 
+        # Via _connection, not the connection property: the property builds a
+        # Connection on first access, so shutting down before a first connect
+        # would construct one -- and raise on an invalid CONNECTION_TYPE --
+        # just to ask whether anything was open.
+        if self._connection is None:
+            return
+
         self._clean_session()
-        if self.connection.connected:
-            await self.connection.close()
+        if self._connection.connected:
+            await self._connection.close()
             logger.info("Disconnected from the Alarm Panel")
 
     async def pause(self):
diff --git a/tests/connection/ip/test_connect_retry.py b/tests/connection/ip/test_connect_retry.py
new file mode 100644
index 00000000..f3689bba
--- /dev/null
+++ b/tests/connection/ip/test_connect_retry.py
@@ -0,0 +1,76 @@
+"""A failed IP connect attempt must not leave its socket behind.
+
+``_try_connect`` stores the protocol as soon as the socket opens and only then
+runs the module handshake. A handshake failure used to leave that socket open
+and unowned: the next attempt overwrote ``_protocol``, and
+``ConnectionProtocol`` deliberately does not close the transport in
+``__del__``. The IP module serves one session at a time, so PAI spent its three
+retries competing with its own orphans -- and it took them back to back, with
+no pause for the module to release the previous session.
+"""
+
+import asyncio
+from unittest.mock import AsyncMock
+
+import pytest
+
+from paradox.connections.ip import connection as ip_connection
+from paradox.connections.ip.commands import IPModuleConnectCommand
+from paradox.connections.ip.connection import LocalIPConnection
+from paradox.connections.protocol_base import ConnectionProtocol
+
+
+@pytest.mark.asyncio
+async def test_each_failed_attempt_is_closed_and_followed_by_a_pause(mocker):
+    connection = LocalIPConnection(host="localhost", port=1000, password="test")
+
+    protocols = []
+    for _ in range(3):
+        protocol = mocker.Mock(spec=ConnectionProtocol)
+        protocol.is_active.return_value = True
+        protocol.close = AsyncMock()
+        protocols.append(protocol)
+
+    create_connection = AsyncMock(side_effect=[(None, p) for p in protocols])
+    mocker.patch.object(
+        asyncio.get_event_loop(), "create_connection", create_connection
+    )
+    mocker.patch.object(
+        IPModuleConnectCommand, "execute", AsyncMock(side_effect=asyncio.TimeoutError)
+    )
+    sleep = mocker.patch.object(ip_connection.asyncio, "sleep", AsyncMock())
+
+    assert await connection.connect() is False
+
+    assert create_connection.await_count == 3
+    for protocol in protocols:
+        protocol.close.assert_awaited_once()
+
+    assert connection._protocol is None
+    assert connection.connected is False
+
+    # Two pauses between three attempts, and none after the last.
+    assert sleep.await_args_list == [mocker.call(ip_connection.CONNECT_RETRY_DELAY)] * 2
+
+
+@pytest.mark.asyncio
+async def test_a_successful_connect_is_not_torn_down_or_delayed(mocker):
+    connection = LocalIPConnection(host="localhost", port=1000, password="test")
+
+    protocol = mocker.Mock(spec=ConnectionProtocol)
+    protocol.is_active.return_value = True
+    protocol.close = AsyncMock()
+
+    mocker.patch.object(
+        asyncio.get_event_loop(),
+        "create_connection",
+        AsyncMock(return_value=(None, protocol)),
+    )
+    mocker.patch.object(IPModuleConnectCommand, "execute", AsyncMock())
+    sleep = mocker.patch.object(ip_connection.asyncio, "sleep", AsyncMock())
+
+    assert await connection.connect() is True
+
+    protocol.close.assert_not_awaited()
+    sleep.assert_not_awaited()
+    assert connection.connected is True
diff --git a/tests/connection/ip/test_stun_session_refresh.py b/tests/connection/ip/test_stun_session_refresh.py
new file mode 100644
index 00000000..7d012701
--- /dev/null
+++ b/tests/connection/ip/test_stun_session_refresh.py
@@ -0,0 +1,106 @@
+"""Renewing the TURN allocation must not block the event loop.
+
+The refresh used to run inline in ``StunIPConnection.write()``, doing blocking
+socket I/O on the event loop. Roughly every 500 s PAI stopped servicing the
+panel, MQTT and every other interface until the TURN server answered -- and
+with no socket timeout, a half-dead control socket stalled it indefinitely.
+Everything then timed out at once, which looks exactly like a panel dropout.
+"""
+
+import asyncio
+import threading
+from unittest.mock import AsyncMock
+
+import pytest
+
+from paradox.connections.ip.connection import StunIPConnection
+from paradox.connections.protocol_base import ConnectionProtocol
+from paradox.exceptions import StunSessionRefreshFailed
+
+
+def _connection(mocker):
+    connection = StunIPConnection(
+        site_id="home", email="em@em.em", panel_serial=None, password="test"
+    )
+    protocol = mocker.Mock(spec=ConnectionProtocol)
+    protocol.is_active.return_value = True
+    protocol.close = AsyncMock()
+    connection._protocol = protocol
+    connection.connected = True
+
+    mocker.patch.object(connection.stun_session, "refresh_required", return_value=True)
+    return connection
+
+
+@pytest.mark.asyncio
+async def test_a_due_refresh_runs_off_the_event_loop(mocker):
+    connection = _connection(mocker)
+
+    started = threading.Event()
+    release = threading.Event()
+    refresh_thread = {}
+
+    def slow_refresh():
+        refresh_thread["ident"] = threading.get_ident()
+        started.set()
+        release.wait(5)
+
+    mocker.patch.object(connection.stun_session, "refresh_session", slow_refresh)
+
+    connection.write(b"payload")
+
+    # The write went out rather than waiting on the TURN server.
+    connection._protocol.send_message.assert_called_once()
+
+    # Let the refresh task start and hand its blocking call to a worker.
+    await asyncio.sleep(0)
+    assert started.wait(5)
+    # The loop is still servicing coroutines while the refresh is blocked.
+    await asyncio.wait_for(asyncio.sleep(0), 1)
+    assert refresh_thread["ident"] != threading.get_ident()
+
+    release.set()
+    await asyncio.wait_for(connection._refresh_task, 5)
+    assert connection.connected is True
+
+
+@pytest.mark.asyncio
+async def test_only_one_refresh_is_in_flight_at_a_time(mocker):
+    connection = _connection(mocker)
+
+    release = threading.Event()
+    calls = []
+
+    def slow_refresh():
+        calls.append(1)
+        release.wait(5)
+
+    mocker.patch.object(connection.stun_session, "refresh_session", slow_refresh)
+
+    connection.write(b"one")
+    first_task = connection._refresh_task
+    connection.write(b"two")
+
+    assert connection._refresh_task is first_task
+
+    release.set()
+    await asyncio.wait_for(first_task, 5)
+    assert calls == [1]
+
+
+@pytest.mark.asyncio
+async def test_a_failed_refresh_drops_the_connection(mocker):
+    connection = _connection(mocker)
+    mocker.patch.object(
+        connection.stun_session,
+        "refresh_session",
+        side_effect=StunSessionRefreshFailed("allocation gone"),
+    )
+
+    connection.write(b"payload")
+    await asyncio.wait_for(connection._refresh_task, 5)
+
+    # The tunnel has stopped forwarding, so the main loop must reconnect
+    # rather than keep writing into it.
+    assert connection.connected is False
+    assert connection._protocol is None
diff --git a/tests/connection/test_connection_close.py b/tests/connection/test_connection_close.py
new file mode 100644
index 00000000..7631b0a3
--- /dev/null
+++ b/tests/connection/test_connection_close.py
@@ -0,0 +1,62 @@
+"""Closing must drop the connection's state even when it fails.
+
+``ConnectionProtocol.close()`` awaits the transport's closed future, which
+carries an *exception* whenever the link died with one -- the normal case when
+closing after a fault. It can also time out waiting for the transport to
+settle. The state reset used to sit after that await, so a raising close left
+``_protocol`` pointing at a dead protocol for the next connect attempt to
+inherit.
+"""
+
+from unittest.mock import AsyncMock
+
+import pytest
+
+from paradox.connections.connection import Connection
+from paradox.connections.protocol_base import ConnectionProtocol
+
+
+class _Connection(Connection):
+    async def connect(self) -> bool:
+        return True
+
+
+def _connected(mocker, close_side_effect=None):
+    connection = _Connection()
+    protocol = mocker.Mock(spec=ConnectionProtocol)
+    protocol.is_active.return_value = True
+    protocol.close = AsyncMock(side_effect=close_side_effect)
+    connection._protocol = protocol
+    connection.connected = True
+    return connection
+
+
+@pytest.mark.asyncio
+async def test_close_resets_state_when_the_protocol_raises(mocker):
+    connection = _connected(mocker, ConnectionResetError("connection reset by peer"))
+
+    with pytest.raises(ConnectionResetError):
+        await connection.close()
+
+    assert connection._protocol is None
+    assert connection.connected is False
+
+
+@pytest.mark.asyncio
+async def test_close_resets_state_on_a_clean_close(mocker):
+    connection = _connected(mocker)
+
+    await connection.close()
+
+    assert connection._protocol is None
+    assert connection.connected is False
+
+
+@pytest.mark.asyncio
+async def test_close_is_idempotent():
+    connection = _Connection()
+
+    await connection.close()
+
+    assert connection._protocol is None
+    assert connection.connected is False
diff --git a/tests/lib/test_stun_client.py b/tests/lib/test_stun_client.py
new file mode 100644
index 00000000..673e8be0
--- /dev/null
+++ b/tests/lib/test_stun_client.py
@@ -0,0 +1,84 @@
+"""STUN runs over TCP, which does not preserve message boundaries.
+
+``receive_response`` used to assert that a single ``recv()`` returned the whole
+message, so ordinary segmentation surfaced as an unexplained connect failure.
+The sockets also had no timeout: a TURN server that stopped answering blocked
+the caller forever, and the session refresh used to run on the event loop.
+"""
+
+import pytest
+
+from paradox.lib import stun
+
+
+class _SegmentedSocket:
+    """Hands back at most ``chunk_size`` bytes per recv(), like a busy link."""
+
+    def __init__(self, payload, chunk_size):
+        self._payload = payload
+        self._chunk_size = chunk_size
+        self.reads = 0
+
+    def recv(self, length):
+        self.reads += 1
+        take = min(length, self._chunk_size)
+        chunk, self._payload = self._payload[:take], self._payload[take:]
+        return chunk
+
+
+def _client(payload, chunk_size):
+    client = stun.StunClient.__new__(stun.StunClient)
+    client.sock = _SegmentedSocket(payload, chunk_size)
+    client.transaction_id = b"T" * 12
+    return client
+
+
+def test_recv_exactly_reassembles_a_segmented_read():
+    client = _client(b"0123456789", chunk_size=3)
+
+    assert client._recv_exactly(10) == b"0123456789"
+    assert client.sock.reads == 4
+
+
+def test_recv_exactly_raises_when_the_peer_closes_early():
+    client = _client(b"012", chunk_size=3)
+
+    with pytest.raises(Exception, match="Connection closed"):
+        client._recv_exactly(10)
+
+
+def test_receive_response_reassembles_a_split_header():
+    header = (
+        stun.BINDING_RESPONSE_SUCCESS  # message type
+        + b"\x00\x00"  # body length: no attributes
+        + stun.MAGIC_COOKIE
+        + b"T" * 12  # transaction id
+    )
+    client = _client(header, chunk_size=7)
+
+    assert client.receive_response() == []
+    # The 20 byte header took three reads; a single recv() would have truncated
+    # it and tripped the old length assertion.
+    assert client.sock.reads == 3
+
+
+def test_receive_response_rejects_a_foreign_transaction_id():
+    header = (
+        stun.BINDING_RESPONSE_SUCCESS
+        + b"\x00\x00"
+        + stun.MAGIC_COOKIE
+        + b"X" * 12  # not the client's transaction id
+    )
+    client = _client(header, chunk_size=20)
+
+    with pytest.raises(Exception, match="invalid transaction id"):
+        client.receive_response()
+
+
+def test_stun_client_bounds_blocking_socket_operations(mocker):
+    sock = mocker.Mock()
+    mocker.patch.object(stun.socket, "socket", return_value=sock)
+
+    stun.StunClient(host="turn.example.com")
+
+    sock.settimeout.assert_called_once_with(stun.STUN_SOCKET_TIMEOUT)
diff --git a/tests/paradox/test_io_timeout_config.py b/tests/paradox/test_io_timeout_config.py
new file mode 100644
index 00000000..f7ce3e84
--- /dev/null
+++ b/tests/paradox/test_io_timeout_config.py
@@ -0,0 +1,46 @@
+"""IO_TIMEOUT has to be read at call time, not frozen into a default argument.
+
+``pai_run`` imports ``paradox.main`` -- and through it ``paradox.paradox``,
+``paradox.lib.handlers`` and the IP connection -- before ``main()`` calls
+``cfg.load()``. A ``timeout=cfg.IO_TIMEOUT`` default argument is evaluated at
+import time, so every request path silently kept the built-in 0.5 s no matter
+what the user configured, and raising IO_TIMEOUT to cope with a slow link did
+nothing at all.
+"""
+
+import asyncio
+from unittest.mock import AsyncMock
+
+import pytest
+
+from paradox.config import config as cfg
+from paradox.lib.handlers import FutureHandler, HandlerRegistry
+from paradox.paradox import Paradox
+
+
+@pytest.mark.asyncio
+async def test_send_wait_uses_the_configured_timeout(mocker, monkeypatch):
+    monkeypatch.setattr(cfg, "IO_TIMEOUT", 7.0)
+
+    alarm = Paradox()
+    alarm._connection = mocker.Mock()
+    alarm._connection.connected = True
+    alarm._connection.wait_for_message = AsyncMock(return_value="reply")
+
+    assert await alarm.send_wait(message=b"x", reply_expected=0x1) == "reply"
+
+    # send_wait allows a reply twice the configured IO timeout.
+    assert alarm._connection.wait_for_message.await_args.kwargs["timeout"] == 14.0
+
+
+@pytest.mark.asyncio
+async def test_wait_until_complete_uses_the_configured_timeout(monkeypatch):
+    monkeypatch.setattr(cfg, "IO_TIMEOUT", 0.01)
+
+    registry = HandlerRegistry()
+
+    with pytest.raises(asyncio.TimeoutError):
+        await registry.wait_until_complete(FutureHandler())
+
+    # The handler is removed even when the wait times out.
+    assert len(registry) == 0
diff --git a/tests/paradox/test_status_poll.py b/tests/paradox/test_status_poll.py
new file mode 100644
index 00000000..79deeb68
--- /dev/null
+++ b/tests/paradox/test_status_poll.py
@@ -0,0 +1,104 @@
+"""The status poll cycle must be bounded and self-contained.
+
+``Paradox.loop`` polls every RAM status address once per ``KEEP_ALIVE_INTERVAL``.
+The requests are serialised behind ``request_lock`` and ``send_wait`` retries
+five times, so an unbounded cycle can outlive the interval by a wide margin --
+publishing no status and counting no missing reply for the whole time.
+
+``asyncio.gather`` makes that worse: it propagates the first failure but leaves
+its siblings running, so the stragglers stay queued on ``request_lock`` and
+collide with the next cycle's requests. One slow address then compounds into
+cross-cycle contention instead of staying in its own cycle.
+"""
+
+import asyncio
+
+import pytest
+
+from paradox.config import config as cfg
+from paradox.exceptions import StatusRequestException
+from paradox.hardware import Panel
+from paradox.hardware.prt3.panel import PRT3Panel
+from paradox.paradox import Paradox
+
+
+class _TenAddressPanel(Panel):
+    status_request_addresses = list(range(10))
+
+
+@pytest.mark.asyncio
+async def test_poll_status_cancels_siblings_when_one_request_fails(mocker):
+    started = asyncio.Event()
+    cancelled = asyncio.Event()
+
+    async def slow_sibling():
+        started.set()
+        try:
+            await asyncio.sleep(30)
+        except asyncio.CancelledError:
+            cancelled.set()
+            raise
+
+    async def failing():
+        await started.wait()
+        raise StatusRequestException("no reply to status request: 1")
+
+    alarm = Paradox()
+    alarm.panel = mocker.Mock(spec=Panel)
+    alarm.panel.get_status_requests.return_value = [slow_sibling(), failing()]
+
+    with pytest.raises(StatusRequestException):
+        await alarm._poll_status()
+
+    await asyncio.wait_for(cancelled.wait(), 1)
+
+
+@pytest.mark.asyncio
+async def test_poll_status_keeps_the_blocks_that_did_parse(mocker):
+    """``handle_status()`` returns None for a block it cannot parse.
+
+    Passing that to ``deep_merge`` raises, which the loop logged as a generic
+    error -- dropping the whole cycle, including the blocks that parsed fine,
+    without counting a missing reply.
+    """
+
+    async def parsed():
+        return {"zone": {1: {"open": True}}}
+
+    async def unparsable():
+        return None
+
+    alarm = Paradox()
+    alarm.panel = mocker.Mock(spec=Panel)
+    alarm.panel.get_status_requests.return_value = [parsed(), unparsable()]
+
+    assert await alarm._poll_status() == {"zone": {1: {"open": True}}}
+
+
+def test_cycle_budget_covers_every_status_address(monkeypatch):
+    monkeypatch.setattr(cfg, "IO_TIMEOUT", 1.0)
+    monkeypatch.setattr(cfg, "KEEP_ALIVE_INTERVAL", 10)
+
+    # Ten addresses, each allowed a reply window plus one retry.
+    assert _TenAddressPanel(core=None).status_cycle_budget == 40.0
+
+
+def test_cycle_budget_never_drops_below_the_keepalive_interval(monkeypatch):
+    monkeypatch.setattr(cfg, "IO_TIMEOUT", 0.1)
+    monkeypatch.setattr(cfg, "KEEP_ALIVE_INTERVAL", 30)
+
+    assert _TenAddressPanel(core=None).status_cycle_budget == 30
+
+
+def test_prt3_cycle_budget_covers_every_area_and_zone(monkeypatch):
+    """PRT3 expands one virtual address into a request per area and zone.
+
+    Sizing its budget from ``status_request_addresses`` -- a single entry --
+    would cancel every cycle on a panel with a realistic zone count.
+    """
+    monkeypatch.setattr(cfg, "IO_TIMEOUT", 0.5)
+    monkeypatch.setattr(cfg, "KEEP_ALIVE_INTERVAL", 10)
+    monkeypatch.setattr(cfg, "PRT3_MAX_AREAS", 8)
+    monkeypatch.setattr(cfg, "PRT3_MAX_ZONES", 96)
+
+    assert PRT3Panel(core=None).status_cycle_budget == 104.0
diff --git a/tests/test_main_uptime.py b/tests/test_main_uptime.py
index 972fd8fd..4004f04d 100644
--- a/tests/test_main_uptime.py
+++ b/tests/test_main_uptime.py
@@ -25,25 +25,26 @@ async def disconnect(self):
         pass
 
 
-async def _nosleep(*args, **kwargs):
-    return None
-
-
-async def run_scripted(script, caplog):
+async def run_scripted(script, caplog, sleeps=None):
     alarm = FakeAlarm(script)
     interface_manager = MagicMock()
     interface_manager.interfaces = []
 
+    async def _record_sleep(delay=None, *args, **kwargs):
+        if sleeps is not None:
+            sleeps.append(delay)
+        return None
+
     clock = [1000.0]
 
     def monotonic():
         clock[0] += 100
         return clock[0]
 
-    with patch.object(
-        pai_main, "InterfaceManager", return_value=interface_manager
-    ), patch.object(pai_main.asyncio, "sleep", new=_nosleep), patch.object(
-        pai_main.time, "monotonic", monotonic
+    with (
+        patch.object(pai_main, "InterfaceManager", return_value=interface_manager),
+        patch.object(pai_main.asyncio, "sleep", new=_record_sleep),
+        patch.object(pai_main.time, "monotonic", monotonic),
     ):
         with caplog.at_level(logging.DEBUG, logger="PAI"):
             await pai_main._run(alarm)
@@ -78,3 +79,17 @@ async def test_banner_reports_version_and_connection(caplog):
 
     assert any("PAI " in m for m in messages)
     assert any("Connection:" in m for m in messages)
+
+
+async def test_reconnect_backoff_grows_and_caps(caplog):
+    """``2 ^ retry`` is XOR, not exponentiation.
+
+    The old expression backed off 3, 0, 1, 6, 7, 4, 5 ... seconds: the second
+    attempt reconnected instantly, the sequence never grew, and it never
+    reached the 30 s cap. The IP module serves one session at a time and needs
+    a moment to release the previous one before it will accept a new one.
+    """
+    sleeps = []
+    await run_scripted(["fail"] * 6 + ["stop"], caplog, sleeps=sleeps)
+
+    assert sleeps == [2, 4, 8, 16, 30, 30]