Harness: no-write guard and stale-container preflight; retire live held-write scenario

Makefile

@@ -33,11 +33,11 @@ lint-dup:
 # --- Dev environment ---
 
 dev:
-	docker compose -f docker-compose.dev.yml up -d
+	docker compose -p exo_pool -f docker-compose.dev.yml up -d
 	python3 scripts/dev-setup.py
 
 stop:
-	docker compose -f docker-compose.dev.yml down
+	docker compose -p exo_pool -f docker-compose.dev.yml down
 
 logs:
 	docker logs ha-exo-pool-dev -f --tail 50

README.md

@@ -218,7 +218,14 @@ Regenerate it after changing `requirements-test.txt` with `make test-lock-regen`
 
 `scripts/verify_outage_reconnect.py` drives the running dev container through
 a simulated WAN outage - only ever against `ha-exo-pool-dev` on port 8125,
-never a live instance. It runs four scenarios:
+never a live instance. It never calls an HA service: every HA API call
+goes through `_ha_request`, which allows only GETs and a POST to the
+config-entry reload path, raising before any request goes out for
+anything else. `Container.exec` (and the netns sidecar's `docker run`)
+carry no such guard - they run whatever command they're given - but the
+only things this script ever asks them to do are read the HA log,
+manage iptables rules, and read/rewrite `/etc/hosts`. It runs four
+scenarios:
 
 - **reconnect-from-connected** (issue #2's actual reproduction): MQTT is
   connected, every one of its actual established peers is blocked (and
@@ -236,6 +243,39 @@ never a live instance. It runs four scenarios:
   code path (setup, not the reconnect chain) - and pins what that does,
   including recovery once the outage clears.
 
+The #12 early-wake-on-reconnect behaviour (a held write retries the instant
+MQTT reconnects rather than waiting out the rest of its cooldown) is unit
+tested in `tests/test_api_write_manager_mqtt_throttle.py` instead of driven
+through this harness - it doesn't need a real device write to prove.
+`_ha_request`'s GET/reload-only allow-list is what keeps this harness from
+ever writing: `tests/test_verify_outage_reconnect.py` proves a write-service
+POST (and PUT/DELETE) never reach `urlopen`, that the script imports none
+of http/http.client/socket/ssl/requests/aiohttp/httpx/urllib3 (a denylist,
+so an unrelated new import like `itertools` doesn't fail it), and that
+`urlopen`/`build_opener`/`Request`/`HTTPConnection`/`HTTPSConnection` are
+named nowhere in the whole module except inside `_ha_request` itself - not
+just inside other functions. These guards catch accidental reintroduction
+of a device write; deliberate evasion (a string-built `getattr`, a
+host-side `subprocess` shelling out to `curl`, a container `exec`) is out
+of scope for them.
+
+Before any scenario runs, `assert_mounted_code_is_loaded()` checks the
+container against this checkout: that its mounted
+`custom_components/exo_pool` resolves to this repo's copy (not some other
+checkout's), and that `State.StartedAt` postdates the newest mtime under
+it. Config-entry reloads re-run setup but never re-import Python modules,
+so a container that predates an edit - or mounts a different checkout
+entirely - keeps running the old code with no error, indistinguishable
+from a real regression until you notice the mismatch. Run this script from
+the same checkout `docker-compose.dev.yml` mounts into `ha-exo-pool-dev`;
+if the mount points elsewhere, repoint it with
+`docker compose -p exo_pool -f docker-compose.dev.yml up -d --force-recreate`
+run from the checkout you want mounted. Restart with
+`docker restart ha-exo-pool-dev` if only the staleness check fires.
+`make dev` always passes `-p exo_pool`, so running it from a worktree
+reuses this same project and config volume instead of standing up a
+second, empty one under the worktree directory's own name.
+
 Two blocking mechanisms, deliberately not unified: reconnect-from-connected
 and interrupt-resume-recovers block by address - read from the container's
 actual established TCP connections (`ss -tn state established`, filtered

scripts/dev-setup.py

@@ -279,7 +279,7 @@ def start_docker() -> None:
 
     print("Starting HA container...")
     subprocess.run(
-        ["docker", "compose", "-f", "docker-compose.dev.yml", "up", "-d"],
+        ["docker", "compose", "-p", "exo_pool", "-f", "docker-compose.dev.yml", "up", "-d"],
         capture_output=True,
     )
 

scripts/verify_outage_reconnect.py

@@ -18,6 +18,7 @@
 import json
 import logging
 import os
+import pathlib
 import re
 import signal
 import subprocess
@@ -26,9 +27,12 @@
 import urllib.error
 import urllib.request
 from dataclasses import dataclass
+from datetime import datetime, timezone
 from typing import Callable
 from urllib.parse import urlparse
 
+_REPO_ROOT = pathlib.Path(__file__).resolve().parent.parent
+
 _LOGGER = logging.getLogger(__name__)
 
 # --- Dev-instance safety -----------------------------------------------
@@ -52,6 +56,14 @@ class NotDevInstanceError(Exception):
     """Raised when the target URL is not confirmed to be the dev container."""
 
 
+class MountedCodeError(Exception):
+    """Raised when a container's mounted exo_pool code isn't this repo's, or
+    predates its own last edit."""
+
+
+MOUNTED_EXO_POOL_DESTINATION = "/config/custom_components/exo_pool"
+
+
 def assert_dev_instance_url(url: str) -> None:
     """Hard-fail unless `url` points at the dev container's port - never the live instance."""
     if urlparse(url).port != DEV_HA_PORT:
@@ -244,6 +256,9 @@ class ScenarioFailure(Exception):
     """Raised when a scenario's assertion doesn't hold."""
 
 
+SCENARIO_EXCEPTIONS = (ScenarioFailure, AssertionError, RuntimeError, urllib.error.URLError, TimeoutError)
+
+
 # --- Container control -----------------------------------------------------
 
 
@@ -270,6 +285,26 @@ def exec(self, cmd: list[str], *, input_text: str | None = None, timeout: int =
             docker_cmd, input=input_text, capture_output=True, text=True, timeout=timeout,
         )
 
+    def started_at(self) -> float:
+        result = self._runner(
+            ["docker", "inspect", "-f", "{{.State.StartedAt}}", self.name],
+            capture_output=True, text=True, timeout=10,
+        )
+        if result.returncode != 0:
+            raise RuntimeError(f"could not inspect {self.name}: {result.stderr.strip()}")
+        return _parse_docker_timestamp(result.stdout.strip())
+
+    def mount_source(self, destination: str) -> str:
+        fmt = f'{{{{range .Mounts}}}}{{{{if eq .Destination "{destination}"}}}}{{{{.Source}}}}{{{{end}}}}{{{{end}}}}'
+        result = self._runner(
+            ["docker", "inspect", "--format", fmt, self.name],
+            capture_output=True, text=True, timeout=10,
+        )
+        source = result.stdout.strip()
+        if result.returncode != 0 or not source:
+            raise RuntimeError(f"{self.name} has no mount at {destination}: {result.stderr.strip()}")
+        return source
+
     def logs_since(self, since_iso: str) -> str:
         """Read HA_LOG_PATH inside the container, filtered to lines since `since_iso`.
 
@@ -288,6 +323,50 @@ def logs_since(self, since_iso: str) -> str:
         return filter_log_lines_since(result.stdout, since_iso)
 
 
+def _parse_docker_timestamp(value: str) -> float:
+    """Parse a `docker inspect` RFC3339 timestamp (nanosecond fraction) to a Unix epoch float."""
+    if "." in value:
+        whole, frac_and_zone = value.split(".", 1)
+        frac = frac_and_zone.rstrip("Z")[:6]
+        value = f"{whole}.{frac}+00:00"
+    else:
+        value = value.rstrip("Z") + "+00:00"
+    return datetime.fromisoformat(value).timestamp()
+
+
+def _epoch_to_iso(epoch: float) -> str:
+    return datetime.fromtimestamp(epoch, tz=timezone.utc).isoformat()
+
+
+EXPECTED_EXO_POOL_SOURCE = _REPO_ROOT / "custom_components" / "exo_pool"
+
+
+def assert_mounted_code_is_loaded(
+    container: Container,
+    mounted_destination: str = MOUNTED_EXO_POOL_DESTINATION,
+    expected_source: pathlib.Path = EXPECTED_EXO_POOL_SOURCE,
+) -> None:
+    """Fail loudly if `container` mounts the wrong checkout, or predates its own mounted code."""
+    started_at = container.started_at()
+    source_dir = pathlib.Path(container.mount_source(mounted_destination))
+    if source_dir.resolve() != expected_source.resolve():
+        raise MountedCodeError(
+            f"{container.name} mounts {source_dir} at {mounted_destination}, not "
+            f"this repo's {expected_source} - point the dev container at this "
+            "checkout before running scenarios."
+        )
+    py_files = sorted(source_dir.glob("*.py"))
+    if not py_files:
+        raise RuntimeError(f"no .py files found under {source_dir}")
+    newest_mtime = max(p.stat().st_mtime for p in py_files)
+    if newest_mtime > started_at:
+        raise MountedCodeError(
+            f"{container.name} started at {_epoch_to_iso(started_at)} but {source_dir} "
+            f"has code modified at {_epoch_to_iso(newest_mtime)} - restart the dev "
+            f"container (`docker restart {container.name}`) before running scenarios."
+        )
+
+
 def should_print_tick(elapsed: float, last_print: float | None, print_interval: float) -> bool:
     """True on the first tick (immediate), or every `print_interval` seconds after that."""
     return last_print is None or elapsed - last_print >= print_interval
@@ -392,7 +471,18 @@ def load_ha_token() -> str:
     )
 
 
+class DisallowedHaRequestError(RuntimeError):
+    """Raised when _ha_request is asked for anything but a GET or a config-entry reload POST."""
+
+
+_ALLOWED_RELOAD_PATH_RE = re.compile(r"/api/config/config_entries/entry/[0-9A-Za-z]+/reload")
+
+
 def _ha_request(method: str, path: str, token: str, data: dict | None = None, timeout: float = 10.0) -> dict | list | None:
+    if method != "GET" and not (method == "POST" and _ALLOWED_RELOAD_PATH_RE.fullmatch(path)):
+        raise DisallowedHaRequestError(
+            f"refusing {method} {path} - only GET and a config-entry reload POST are allowed"
+        )
     url = f"{HA_URL}{path}"
     body = json.dumps(data).encode() if data is not None else None
     req = urllib.request.Request(
@@ -1243,6 +1333,12 @@ def main() -> int:
         print(f"FATAL: container {CONTAINER_NAME!r} is not running. Run `make dev` first.")
         return 1
 
+    try:
+        assert_mounted_code_is_loaded(container)
+    except (MountedCodeError, RuntimeError) as e:
+        print(f"FATAL: {e}")
+        return 1
+
     token = load_ha_token()
 
     try:
@@ -1285,7 +1381,7 @@ def _recover_between_scenarios() -> None:
         try:
             scenario_baseline(token, entry_id, mqtt_entity, container)
             results["baseline"] = "PASS"
-        except (ScenarioFailure, AssertionError, RuntimeError) as e:
+        except SCENARIO_EXCEPTIONS as e:
             results["baseline"] = f"FAIL: {e}"
         _recover_between_scenarios()
 
@@ -1294,7 +1390,7 @@ def _recover_between_scenarios() -> None:
             results["reconnect_from_connected"] = "PASS" if outcome else (
                 "SKIP (no NET_ADMIN)" if outcome is None else "FAIL"
             )
-        except (ScenarioFailure, AssertionError, RuntimeError) as e:
+        except SCENARIO_EXCEPTIONS as e:
             results["reconnect_from_connected"] = f"FAIL: {e}"
         _recover_between_scenarios()
 
@@ -1303,7 +1399,7 @@ def _recover_between_scenarios() -> None:
             results["interrupt_resume_recovers"] = "PASS" if outcome else (
                 "SKIP (no NET_ADMIN)" if outcome is None else "FAIL"
             )
-        except (ScenarioFailure, AssertionError, RuntimeError) as e:
+        except SCENARIO_EXCEPTIONS as e:
             results["interrupt_resume_recovers"] = f"FAIL: {e}"
         _recover_between_scenarios()
 
@@ -1312,7 +1408,7 @@ def _recover_between_scenarios() -> None:
             results["setup_under_outage"] = "PASS" if outcome else (
                 "SKIP (no NET_ADMIN)" if outcome is None else "FAIL"
             )
-        except (ScenarioFailure, AssertionError, RuntimeError) as e:
+        except SCENARIO_EXCEPTIONS as e:
             results["setup_under_outage"] = f"FAIL: {e}"
         _recover_between_scenarios()
 
@@ -1324,7 +1420,7 @@ def _recover_between_scenarios() -> None:
                 results["watchdog"] = "PASS" if outcome else (
                     "SKIP (no NET_ADMIN)" if outcome is None else "FAIL"
                 )
-            except (ScenarioFailure, AssertionError, RuntimeError) as e:
+            except SCENARIO_EXCEPTIONS as e:
                 results["watchdog"] = f"FAIL: {e}"
     finally:
         teardown.run()

tests/test_api_write_manager_mqtt_throttle.py

@@ -428,6 +428,7 @@ async def fake_sleep(seconds):
 
 
 class TestStateChangedCallbackWiring:
+    @pytest.mark.timeout(5)
     async def test_reconnect_callback_wakes_a_held_write_before_the_cooldown_ends(
         self, hass, entry, disconnected_mqtt, monkeypatch, fake_clock,
         post_write_cooldown_seconds,