Keep WiFi test recovery retrying until the original network returns

.github/workflows/run-unit-tests.yml

@@ -274,6 +274,7 @@ jobs:
         run: |
           python3 -B test/test_official_web_console.py -v
           python3 -B test/test_hil_linux_web_serial.py -v
+          python3 -B test/test_hil_slowfi_guard.py -v
 
       - name: Verify official setup GUI initial time command deadline
         run: python3 -B test/test_official_config_gui.py -v

test/README.md

@@ -92,6 +92,33 @@ python3 -B test/test_hil_linux_web_serial.py -v
 
 See [the scoped G2 comparison and Chromium source evidence](../docs/research/linux_chromium_web_serial_tty.md).
 
+### Timed WiFi return guard
+
+The host-only SlowFi guard tests use fake NetworkManager responses and local
+locks. They cover recurring recovery after more than five failures, bounded
+attempts, exact original UUID plus connected-state proof, overlapping attempts,
+cleanup refusal while the original network is unproven, owned-profile collisions,
+and timer-stop side effects with uncertain results. They do not test a real
+systemd timer, WiFi association, a Pi, or a radio:
+
+```sh
+python3 -B test/test_hil_slowfi_guard.py -v
+```
+
+The helper's CLI defaults to a plan and makes no NetworkManager/systemd calls:
+
+```sh
+python3 -B -m tools.hil.slowfi_guard --config /path/to/reviewed-config.json
+```
+
+A future hardware test must separately bind and stage the reviewed host config,
+use a fresh owned unit identity, and prove the actual first timer deadline before
+changing the Pi's network. The rendered guard first attempts the original UUID
+after 660 seconds, then retries 30 seconds after each bounded attempt until the
+original UUID is connected. It never edits the original profile. Live recovery
+and cleanup require explicit authorization, root-owned staged files and host
+identity checks; cleanup also refuses unless the original network is proven.
+
 The bootloader-version regression compiles the production reader with a C++17
 `g++` (or `CXX`) compiler. Its offline cases include the MeshTower V2 SD 2.4.6
 missing-UF2-text failure, OTAFIX metadata versus cached base-version precedence,

test/test_hil_slowfi_guard.py

@@ -0,0 +1,366 @@
+"""Pure fake-NM tests. No NetworkManager/systemd/SSH/hardware calls."""
+import copy
+import contextlib
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+import importlib.util
+
+ROOT = Path(__file__).resolve().parents[1]
+SPEC = importlib.util.spec_from_file_location(
+    "hil_slowfi_guard", ROOT / "tools/hil/slowfi_guard.py")
+guard = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(guard)
+
+ORIGINAL = "11111111-1111-4111-8111-111111111111"
+OWNED = "22222222-2222-4222-8222-222222222222"
+OTHER = "33333333-3333-4333-8333-333333333333"
+STEM = "meshcore-slowfi-return-synthetic-20261007"
+CONFIG = {
+    "schema": 1, "original_uuid": ORIGINAL,
+    "expected_hostname": "synthetic-no-hardware-host",
+    "expected_machine_id_sha256": "1" * 64, "interface": "wlan0",
+    "unit_stem": STEM, "guard_dir": "/run/" + STEM,
+    "python_path": "/usr/bin/python3",
+    "helper_path": "/run/" + STEM + "/slowfi_guard.py",
+    "config_path": "/run/" + STEM + "/config.json",
+    "live_execution_authorized": False,
+    "owned_profiles": [{"uuid": OWNED, "id": "meshcore-owned-ap-synthetic-20261007"}],
+}
+
+
+class Clock:
+    def __init__(self):
+        self.now = 1000.0
+
+    def __call__(self):
+        return self.now
+
+
+class FakeNM:
+    def __init__(self, clock=None, failures=0):
+        self.clock = clock or Clock()
+        self.failures = failures
+        self.attempts = 0
+        self.current = {"uuid": OWNED, "state": 100}
+        self.timer_active = True
+        self.stop_ok = True
+        self.stop_side_effect = False
+        self.timer_state_unknown = False
+        self.state_sequence = []
+        self.calls = []
+        self.activation_return = None
+        self.activation_state = None
+        self.names = {OWNED: CONFIG["owned_profiles"][0]["id"], ORIGINAL: "original-profile"}
+        self.timeouts = False
+
+    def state(self, interface, timeout):
+        self.calls.append(("state", interface, timeout))
+        self.clock.now += timeout if self.timeouts else 0.1
+        if self.timeouts:
+            return None
+        if self.state_sequence:
+            return copy.deepcopy(self.state_sequence.pop(0))
+        return copy.deepcopy(self.current)
+
+    def activate(self, original_uuid, timeout):
+        self.calls.append(("activate", original_uuid, timeout))
+        self.attempts += 1
+        self.clock.now += timeout if self.timeouts else 0.1
+        if self.timeouts:
+            return False
+        if self.activation_state is not None:
+            self.current = copy.deepcopy(self.activation_state)
+        elif self.attempts > self.failures:
+            self.current = {"uuid": original_uuid, "state": 100}
+        return self.activation_return if self.activation_return is not None else self.attempts > self.failures
+
+    def stop_timer(self, timer, timeout):
+        self.calls.append(("stop_timer", timer, timeout))
+        self.clock.now += 0.1
+        if self.stop_ok or self.stop_side_effect:
+            self.timer_active = False
+        return self.stop_ok
+
+    def timer_state(self, timer, timeout):
+        self.calls.append(("timer_state", timer, timeout))
+        self.clock.now += 0.1
+        return None if self.timer_state_unknown else "active" if self.timer_active else "inactive"
+
+    def stop_owned_units(self, timer, service, timeout):
+        self.calls.append(("stop_owned_units", timer, service, timeout))
+        self.timer_active = False
+        return True
+
+    def profile_name(self, value, timeout):
+        self.calls.append(("profile_name", value, timeout))
+        return self.names.get(value)
+
+    def delete_profile(self, value, timeout):
+        self.calls.append(("delete_profile", value, timeout))
+        del self.names[value]
+        return True
+
+
+class GuardTests(unittest.TestCase):
+    def setUp(self):
+        self.config = copy.deepcopy(CONFIG)
+
+    def run_attempt(self, backend):
+        return guard.attempt(self.config, backend, clock=backend.clock)
+
+    def run_cleanup(self, backend, acquired=True):
+        return guard.explicit_owned_cleanup(self.config, backend, clock=backend.clock,
+            lock_context=lambda _: contextlib.nullcontext(acquired))
+
+    def test_more_than_five_transient_failures_then_connected_stops_timer(self):
+        backend = FakeNM(failures=8)
+        receipts = []
+        for number in range(9):
+            receipts.append(self.run_attempt(backend))
+            if number < 8:
+                self.assertTrue(backend.timer_active)
+                self.assertTrue(receipts[-1]["recurring_retry_retained"])
+                backend.clock.now += guard.RETRY_DELAY_SECONDS
+        self.assertEqual(backend.attempts, 9)
+        self.assertTrue(receipts[-1]["original_uuid_connected"])
+        self.assertTrue(receipts[-1]["timer_stopped"])
+        self.assertFalse(backend.timer_active)
+
+    def test_persistent_failures_leave_unlimited_future_attempts(self):
+        backend = FakeNM(failures=10**9)
+        for _ in range(100):
+            result = self.run_attempt(backend)
+            self.assertFalse(result["original_uuid_connected"])
+            self.assertTrue(result["recurring_retry_retained"])
+            self.assertTrue(backend.timer_active)
+            backend.clock.now += guard.RETRY_DELAY_SECONDS
+        self.assertEqual(backend.attempts, 100)
+        self.assertFalse(any(row[0] == "stop_timer" for row in backend.calls))
+
+    def test_return_zero_is_not_association_proof(self):
+        backend = FakeNM(failures=10)
+        backend.activation_return = True
+        result = self.run_attempt(backend)
+        self.assertTrue(result["nmcli_activation_returned_success"])
+        self.assertFalse(result["original_uuid_connected"])
+        self.assertTrue(backend.timer_active)
+
+    def test_wrong_uuid_or_nonconnected_state_never_stops_timer(self):
+        for state in ({"uuid": OTHER, "state": 100}, {"uuid": ORIGINAL, "state": 70}, {"uuid": ORIGINAL, "state": 30}):
+            with self.subTest(state=state):
+                backend = FakeNM()
+                backend.activation_state = state
+                result = self.run_attempt(backend)
+                self.assertFalse(result["original_uuid_connected"])
+                self.assertTrue(backend.timer_active)
+
+    def test_already_original_connected_has_no_nm_activation(self):
+        backend = FakeNM()
+        backend.current = {"uuid": ORIGINAL, "state": 100}
+        result = self.run_attempt(backend)
+        self.assertTrue(result["timer_stopped"])
+        self.assertEqual(backend.attempts, 0)
+
+    def test_stop_timer_failure_keeps_recurring_retry(self):
+        backend = FakeNM()
+        backend.stop_ok = False
+        result = self.run_attempt(backend)
+        self.assertTrue(result["original_uuid_connected"])
+        self.assertFalse(result["timer_stopped"])
+        self.assertTrue(result["recurring_retry_retained"])
+
+    def test_stop_timer_side_effect_then_failure_does_not_claim_retry_retained(self):
+        backend = FakeNM()
+        backend.stop_ok = False
+        backend.stop_side_effect = True
+        result = self.run_attempt(backend)
+        self.assertFalse(result["timer_stop_returned_success"])
+        self.assertTrue(result["timer_stopped"])
+        self.assertFalse(result["recurring_retry_retained"])
+        self.assertEqual(result["timer_state_after_stop"], "inactive")
+
+    def test_stop_timer_state_unknown_does_not_claim_retry_retained(self):
+        backend = FakeNM()
+        backend.timer_state_unknown = True
+        result = self.run_attempt(backend)
+        self.assertTrue(result["timer_stop_returned_success"])
+        self.assertIsNone(result["recurring_retry_retained"])
+        self.assertFalse(result["timer_stopped"])
+
+    def test_stop_exception_does_not_claim_retry_retained(self):
+        backend = FakeNM()
+        with patch.object(backend, "stop_timer", side_effect=TimeoutError("secret")):
+            result = self.run_attempt(backend)
+        self.assertIsNone(result["recurring_retry_retained"])
+        self.assertFalse(result["timer_stopped"])
+
+    def test_all_subprocess_timeouts_stay_under_35_seconds(self):
+        backend = FakeNM()
+        backend.timeouts = True
+        result = self.run_attempt(backend)
+        self.assertLessEqual(result["elapsed_seconds"], guard.ATTEMPT_BUDGET_SECONDS)
+        self.assertTrue(result["recurring_retry_retained"])
+        self.assertEqual([x[-1] for x in backend.calls], [3, 24, 3])
+
+    def test_nonblocking_lock_skips_overlap_then_allows_next_invocation(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "guard.lock"
+            with guard.owned_lock(path) as first:
+                self.assertTrue(first)
+                with guard.owned_lock(path) as second:
+                    self.assertFalse(second)
+            with guard.owned_lock(path) as third:
+                self.assertTrue(third)
+
+    def test_lock_refuses_symlink(self):
+        with tempfile.TemporaryDirectory() as directory:
+            actual = Path(directory) / "actual"
+            actual.touch()
+            alias = Path(directory) / "alias"
+            alias.symlink_to(actual)
+            with self.assertRaises(OSError):
+                with guard.owned_lock(alias):
+                    pass
+
+    def test_owned_cleanup_uses_only_captured_profile_and_units(self):
+        backend = FakeNM()
+        backend.current = {"uuid": ORIGINAL, "state": 100}
+        result = self.run_cleanup(backend)
+        self.assertTrue(result["ok"])
+        self.assertTrue(result["original_profile_untouched"])
+        self.assertIn(ORIGINAL, backend.names)
+        self.assertNotIn(OWNED, backend.names)
+        self.assertEqual([x[1] for x in backend.calls if x[0] == "delete_profile"], [OWNED])
+        stopped = [row for row in backend.calls if row[0] == "stop_owned_units"]
+        self.assertEqual(stopped[0][:3], ("stop_owned_units", STEM + ".timer", STEM + ".service"))
+        self.assertEqual([row[0] for row in backend.calls[:4]], ["state", "state", "profile_name", "state"])
+
+    def test_owned_cleanup_refuses_changed_profile_name(self):
+        backend = FakeNM()
+        backend.current = {"uuid": ORIGINAL, "state": 100}
+        backend.names[OWNED] = "unowned-existing-profile"
+        result = self.run_cleanup(backend)
+        self.assertTrue(result["cleanup_refused"])
+        self.assertTrue(result["no_mutation"])
+        self.assertTrue(backend.timer_active)
+        self.assertEqual(result["reason"], "owned_profile_identity_changed")
+        self.assertFalse(any(x[0] == "delete_profile" for x in backend.calls))
+
+    def test_cleanup_wrong_uuid_connecting_or_inconclusive_keeps_guard_running(self):
+        for state in ({"uuid": OTHER, "state": 100}, {"uuid": ORIGINAL, "state": 70}, None):
+            with self.subTest(state=state):
+                backend = FakeNM()
+                backend.current = state
+                result = self.run_cleanup(backend)
+                self.assertTrue(result["cleanup_refused"])
+                self.assertTrue(result["no_mutation"])
+                self.assertTrue(backend.timer_active)
+                self.assertEqual([row[0] for row in backend.calls], ["state"])
+
+    def test_cleanup_association_lost_under_lock_refuses_without_mutation(self):
+        for state in ({"uuid": OWNED, "state": 100}, {"uuid": ORIGINAL, "state": 70}, None):
+            with self.subTest(state=state):
+                backend = FakeNM()
+                backend.state_sequence = [{"uuid": ORIGINAL, "state": 100}, state]
+                result = self.run_cleanup(backend)
+                self.assertTrue(result["cleanup_refused"])
+                self.assertTrue(result["no_mutation"])
+                self.assertTrue(backend.timer_active)
+                self.assertEqual([row[0] for row in backend.calls], ["state", "state"])
+
+    def test_later_ownership_collision_prevents_all_cleanup_mutations(self):
+        self.config["owned_profiles"].append({"uuid": OTHER, "id": "meshcore-owned-second-synthetic-20261007"})
+        backend = FakeNM()
+        backend.current = {"uuid": ORIGINAL, "state": 100}
+        backend.names[OTHER] = "unowned-existing-profile"
+        result = self.run_cleanup(backend)
+        self.assertTrue(result["cleanup_refused"])
+        self.assertTrue(result["no_mutation"])
+        self.assertTrue(backend.timer_active)
+        self.assertIn(OWNED, backend.names)
+        self.assertFalse(any(row[0] in {"stop_owned_units", "delete_profile"} for row in backend.calls))
+
+    def test_cleanup_lost_before_stop_refuses_without_mutation(self):
+        backend = FakeNM()
+        backend.state_sequence = [{"uuid": ORIGINAL, "state": 100}, {"uuid": ORIGINAL, "state": 100}, {"uuid": ORIGINAL, "state": 70}]
+        result = self.run_cleanup(backend)
+        self.assertTrue(result["cleanup_refused"])
+        self.assertTrue(result["no_mutation"])
+        self.assertTrue(backend.timer_active)
+        self.assertFalse(any(row[0] in {"stop_owned_units", "delete_profile"} for row in backend.calls))
+
+    def test_cleanup_busy_lock_does_not_stop_any_unit_or_delete_profile(self):
+        backend = FakeNM()
+        backend.current = {"uuid": ORIGINAL, "state": 100}
+        result = self.run_cleanup(backend, acquired=False)
+        self.assertTrue(result["cleanup_refused"])
+        self.assertTrue(result["no_mutation"])
+        self.assertTrue(backend.timer_active)
+        self.assertEqual([row[0] for row in backend.calls], ["state"])
+
+    def test_cleanup_state_error_does_not_stop_guard(self):
+        backend = FakeNM()
+        with patch.object(backend, "state", side_effect=RuntimeError("secret")):
+            result = self.run_cleanup(backend)
+        self.assertTrue(result["cleanup_refused"])
+        self.assertTrue(result["no_mutation"])
+        self.assertTrue(backend.timer_active)
+        self.assertNotIn("secret", str(result))
+
+    def test_original_profile_cannot_be_declared_owned(self):
+        self.config["owned_profiles"][0]["uuid"] = ORIGINAL
+        with self.assertRaisesRegex(ValueError, "original_profile"):
+            guard.validate(self.config)
+
+    def test_invalid_unit_or_path_rejected(self):
+        for key, value in (("unit_stem", "NetworkManager"), ("helper_path", "/tmp/unowned.py")):
+            with self.subTest(key=key):
+                config = copy.deepcopy(CONFIG)
+                config[key] = value
+                with self.assertRaises(ValueError):
+                    guard.validate(config)
+
+    def test_actual_backend_command_builder_never_edits_original_profile(self):
+        backend = guard.LiveBackend.__new__(guard.LiveBackend)
+        backend.interface = "wlan0"
+        with patch.object(backend, "call", return_value=(0, "SECRET-NOT-LOGGED")) as command:
+            self.assertTrue(backend.activate(ORIGINAL, 25))
+            command.assert_called_once_with([guard.NMCLI, "--wait", "25", "connection", "up", "uuid", ORIGINAL, "ifname", "wlan0"], 25)
+        with patch.object(backend, "call", return_value=(0, "")) as command:
+            self.assertTrue(backend.stop_timer(STEM + ".timer", 1))
+            command.assert_called_once_with([guard.SYSTEMCTL, "stop", STEM + ".timer"], 1)
+
+    def test_backend_exception_is_safe_and_does_not_disarm_timer(self):
+        backend = FakeNM()
+        with patch.object(backend, "state", side_effect=RuntimeError("SECRET-not-for-receipt")):
+            result = self.run_attempt(backend)
+        self.assertEqual(result["errors"], ["RuntimeError"])
+        self.assertTrue(result["recurring_retry_retained"])
+        self.assertNotIn("SECRET", str(result))
+        self.assertTrue(backend.timer_active)
+
+    def test_parse_exact_uuid_and_connected_state(self):
+        self.assertEqual(guard.parse_state(ORIGINAL + "\n100 (connected)\n"), {"uuid": ORIGINAL, "state": 100})
+        self.assertIsNone(guard.parse_state("--\n30 (disconnected)\n"))
+        self.assertIsNone(guard.parse_state(ORIGINAL + "\n100\nextra"))
+
+    def test_plan_does_not_construct_live_backend_or_run_command(self):
+        with patch.object(guard.subprocess, "run", side_effect=AssertionError("unexpected command")), patch.object(guard, "LiveBackend", side_effect=AssertionError("unexpected live backend")):
+            result = guard.plan(self.config)
+        self.assertTrue(result["prepared_only"])
+        files = result["unit_files"]
+        timer, service = files[STEM + ".timer"], files[STEM + ".service"]
+        self.assertIn("OnActiveSec=660s", timer)
+        self.assertIn("OnUnitInactiveSec=30s", timer)
+        self.assertIn("AccuracySec=1s", timer)
+        self.assertIn("TimeoutStartSec=35s", service)
+        self.assertIn("RemainAfterExit=no", service)
+        self.assertNotIn("Restart=", service)
+        self.assertNotIn("connection modify", timer + service)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

tools/hil/slowfi_guard.py

@@ -0,0 +1,326 @@
+#!/usr/bin/env python3
+"""Future host recovery helper. Default is a read-only plan; not staged/live-tested."""
+from __future__ import annotations
+
+import argparse
+import contextlib
+import fcntl
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import socket
+import subprocess
+import time
+import uuid
+
+FIRST_DELAY_SECONDS = 660
+RETRY_DELAY_SECONDS = 30
+# The worker has32s for subprocesses; the service enforces35s including startup.
+ATTEMPT_BUDGET_SECONDS = 32
+CLEANUP_BUDGET_SECONDS = 75
+NMCLI = "/usr/bin/nmcli"
+SYSTEMCTL = "/usr/bin/systemctl"
+
+
+def require(condition, code):
+    if not condition:
+        raise ValueError(code)
+
+
+def valid_uuid(value):
+    try:
+        return str(uuid.UUID(value)) == value
+    except (ValueError, TypeError, AttributeError):
+        return False
+
+
+def validate(config):
+    require(config.get("schema") == 1, "schema_mismatch")
+    require(valid_uuid(config.get("original_uuid")), "original_uuid_unbound")
+    require(bool(re.fullmatch(r"[a-zA-Z0-9_.-]+", config.get("expected_hostname", ""))), "hostname_unbound")
+    require(bool(re.fullmatch(r"[a-f0-9]{64}", config.get("expected_machine_id_sha256", ""))), "machine_id_unbound")
+    require(bool(re.fullmatch(r"[a-zA-Z0-9_.-]{1,15}", config.get("interface", ""))), "interface_invalid")
+    stem = config.get("unit_stem", "")
+    require(bool(re.fullmatch(r"meshcore-slowfi-return-[a-z0-9-]{8,80}", stem)), "owned_unit_invalid")
+    root = config.get("guard_dir", "")
+    require(root == "/run/" + stem, "owned_guard_directory_invalid")
+    for key, name in (("helper_path", "slowfi_guard.py"), ("config_path", "config.json")):
+        require(config.get(key) == root + "/" + name, "owned_path_invalid")
+    require(config.get("python_path") == "/usr/bin/python3", "python_path_invalid")
+    require(type(config.get("live_execution_authorized")) is bool, "live_authorization_invalid")
+    owned = config.get("owned_profiles", [])
+    require(isinstance(owned, list) and len(owned) <= 4, "owned_profiles_invalid")
+    seen = set()
+    for item in owned:
+        require(isinstance(item, dict) and set(item) == {"uuid", "id"}, "owned_profile_invalid")
+        require(valid_uuid(item["uuid"]) and item["uuid"] != config["original_uuid"], "original_profile_must_not_be_owned")
+        require(item["uuid"] not in seen, "duplicate_owned_profile")
+        require(bool(re.fullmatch(r"meshcore-[a-z0-9-]{8,100}", item["id"])), "owned_profile_name_invalid")
+        seen.add(item["uuid"])
+    return config
+
+
+def render_units(config):
+    validate(config)
+    stem = config["unit_stem"]
+    timer = f"""[Unit]
+Description=Return to original WiFi until connected ({stem})
+
+[Timer]
+OnActiveSec={FIRST_DELAY_SECONDS}s
+OnUnitInactiveSec={RETRY_DELAY_SECONDS}s
+AccuracySec=1s
+RandomizedDelaySec=0
+Unit={stem}.service
+RemainAfterElapse=yes
+"""
+    service = f"""[Unit]
+Description=One bounded original WiFi recovery attempt ({stem})
+StartLimitIntervalSec=0
+
+[Service]
+Type=oneshot
+RemainAfterExit=no
+User=root
+ExecStart={config['python_path']} {config['helper_path']} attempt --config {config['config_path']} --allow-live-host-recovery
+TimeoutStartSec=35s
+TimeoutStopSec=1s
+KillMode=control-group
+SendSIGKILL=yes
+UMask=0077
+"""
+    return {stem + ".timer": timer, stem + ".service": service}
+
+
+def parse_state(output):
+    lines = output.strip().splitlines()
+    if len(lines) != 2 or not valid_uuid(lines[0].strip()):
+        return None
+    match = re.fullmatch(r"(\d+)(?:\s+\([^\r\n]*\))?", lines[1].strip())
+    if not match:
+        return None
+    return {"uuid": lines[0].strip(), "state": int(match[1])}
+
+
+class LiveBackend:
+    """No instantiation from plan/render/tests; commands never return raw output."""
+    def __init__(self, config):
+        require(config["live_execution_authorized"], "live_execution_not_authorized")
+        require(os.geteuid() == 0, "live_requires_root")
+        require(socket.gethostname() == config["expected_hostname"], "wrong_host")
+        machine_id = Path("/etc/machine-id").read_text().strip().encode()
+        require(hashlib.sha256(machine_id).hexdigest() == config["expected_machine_id_sha256"], "wrong_machine_id")
+        root = Path(config["guard_dir"])
+        stat = root.stat()
+        require(root.is_dir() and not root.is_symlink() and stat.st_uid == 0 and not stat.st_mode & 0o022, "guard_directory_not_owned")
+        for key in ("config_path", "helper_path"):
+            item = Path(config[key])
+            stat = item.stat()
+            require(item.is_file() and not item.is_symlink() and stat.st_uid == 0 and not stat.st_mode & 0o022, "live_file_not_owned")
+        self.interface = config["interface"]
+
+    @staticmethod
+    def call(argv, timeout):
+        try:
+            value = subprocess.run(argv, capture_output=True, text=True, timeout=timeout, check=False, stdin=subprocess.DEVNULL)
+            return value.returncode, value.stdout
+        except subprocess.TimeoutExpired:
+            return None, ""
+
+    def state(self, interface, timeout):
+        rc, output = self.call([NMCLI, "--terse", "--escape", "no", "--get-values", "GENERAL.CON-UUID,GENERAL.STATE", "device", "show", interface], timeout)
+        return parse_state(output) if rc == 0 else None
+
+    def activate(self, original_uuid, timeout):
+        rc, _ = self.call([NMCLI, "--wait", "25", "connection", "up", "uuid", original_uuid, "ifname", self.interface], timeout)
+        return rc == 0
+
+    def stop_timer(self, unit, timeout):
+        rc, _ = self.call([SYSTEMCTL, "stop", unit], timeout)
+        return rc == 0
+
+    def timer_state(self, unit, timeout):
+        rc, output = self.call([SYSTEMCTL, "show", "--property=ActiveState", "--value", unit], timeout)
+        state = output.strip()
+        return state if rc == 0 and state in {"active", "inactive", "failed", "activating", "deactivating"} else None
+
+    def profile_name(self, profile_uuid, timeout):
+        rc, output = self.call([NMCLI, "--get-values", "connection.id", "connection", "show", "uuid", profile_uuid], timeout)
+        return output.strip() if rc == 0 else None
+
+    def delete_profile(self, profile_uuid, timeout):
+        rc, _ = self.call([NMCLI, "--wait", "5", "connection", "delete", "uuid", profile_uuid], timeout)
+        return rc == 0
+
+    def stop_owned_units(self, timer, service, timeout):
+        rc, _ = self.call([SYSTEMCTL, "stop", timer, service], timeout)
+        return rc == 0
+
+
+@contextlib.contextmanager
+def owned_lock(path, expected_uid=None):
+    fd = os.open(path, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
+    try:
+        stat = os.fstat(fd)
+        require(stat.st_nlink == 1 and (expected_uid is None or stat.st_uid == expected_uid), "lock_not_owned")
+        try:
+            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError:
+            yield False
+            return
+        try:
+            yield True
+        finally:
+            fcntl.flock(fd, fcntl.LOCK_UN)
+    finally:
+        os.close(fd)
+
+
+def attempt(config, backend, *, clock=time.monotonic):
+    """Single bounded invocation; no sleep, global retry cap, or profile edit."""
+    validate(config)
+    begun = clock()
+    deadline = begun + ATTEMPT_BUDGET_SECONDS
+    report = {"schema": 1, "original_uuid_connected": False, "activation_attempted": False,
+              "timer_stopped": False, "recurring_retry_retained": True,
+              "attempt_budget_seconds": ATTEMPT_BUDGET_SECONDS, "errors": []}
+
+    def bounded(limit):
+        left = deadline - clock()
+        if left <= 0:
+            raise TimeoutError("attempt_deadline")
+        return min(limit, left)
+
+    def connected(state):
+        return bool(state and state["uuid"] == config["original_uuid"] and state["state"] == 100)
+
+    try:
+        state = backend.state(config["interface"], bounded(3))
+        if not connected(state):
+            report["activation_attempted"] = True
+            # Reserve3s for association and1s each for stop and state proof.
+            allowance = min(24, deadline - clock() - 5)
+            if allowance <= 0:
+                raise TimeoutError("no_activation_budget")
+            report["nmcli_activation_returned_success"] = backend.activate(config["original_uuid"], allowance)
+            state = backend.state(config["interface"], bounded(3))
+        report["original_uuid_connected"] = connected(state)
+        if report["original_uuid_connected"]:
+            timer = config["unit_stem"] + ".timer"
+            report["recurring_retry_retained"] = None
+            report["timer_state_after_stop"] = None
+            report["timer_stop_returned_success"] = backend.stop_timer(timer, bounded(1))
+            # Stop failure/timeout can occur after the side effect. Read the
+            # exact unit rather than infer timer activity from the return code.
+            report["timer_state_after_stop"] = backend.timer_state(timer, bounded(1))
+            report["timer_stopped"] = report["timer_state_after_stop"] == "inactive"
+            report["recurring_retry_retained"] = (
+                True if report["timer_state_after_stop"] == "active" else
+                False if report["timer_state_after_stop"] in {"inactive", "failed"} else None)
+    except Exception as error:
+        # Never store command output, SSIDs, credentials, or exception messages.
+        report["errors"].append(type(error).__name__)
+    report["elapsed_seconds"] = round(clock() - begun, 6)
+    return report
+
+
+def explicit_owned_cleanup(config, backend, *, clock=time.monotonic, lock_context=None):
+    """Refuse cleanup while original WiFi is unproven; one bounded lock owner."""
+    validate(config)
+    deadline = clock() + CLEANUP_BUDGET_SECONDS
+    stem = config["unit_stem"]
+    report = {"ok": False, "cleanup_refused": False, "no_mutation": True,
+              "original_profile_untouched": True, "owned_profiles": [],
+              "cleanup_budget_seconds": CLEANUP_BUDGET_SECONDS}
+
+    def bounded(limit):
+        left = deadline - clock()
+        if left <= 0:
+            raise TimeoutError("cleanup_deadline")
+        return min(left, limit)
+
+    def original_connected():
+        state = backend.state(config["interface"], bounded(3))
+        return bool(state and state["uuid"] == config["original_uuid"] and state["state"] == 100)
+
+    def refuse(code):
+        report.update({"cleanup_refused": True, "reason": code})
+        return report
+
+    try:
+        if not original_connected():
+            return refuse("original_wifi_not_connected_before_lock")
+        context = lock_context or (lambda path: owned_lock(path, expected_uid=0))
+        with context(config["guard_dir"] + "/attempt.lock") as acquired:
+            if not acquired:
+                return refuse("cleanup_attempt_lock_busy")
+            if not original_connected():
+                return refuse("original_wifi_not_connected_under_lock")
+            # Check all ownership before any mutation, avoiding partial cleanup
+            # for a known collision in a later owned-profile entry.
+            for item in config["owned_profiles"]:
+                if backend.profile_name(item["uuid"], bounded(3)) != item["id"]:
+                    return refuse("owned_profile_identity_changed")
+            if not original_connected():
+                return refuse("original_wifi_not_connected_before_unit_stop")
+            report["no_mutation"] = False
+            require(backend.stop_owned_units(stem + ".timer", stem + ".service", bounded(5)), "owned_units_stop_failed")
+            report["owned_units_stopped"] = True
+            for item in config["owned_profiles"]:
+                # Recheck capturedUUID/name immediately before deletion too.
+                require(backend.profile_name(item["uuid"], bounded(3)) == item["id"], "owned_profile_identity_changed")
+                require(backend.delete_profile(item["uuid"], bounded(7)), "owned_profile_delete_failed")
+                report["owned_profiles"].append({"owned_profile_deleted": True})
+            report["ok"] = True
+    except Exception as error:
+        report["error_type"] = type(error).__name__
+        if report["no_mutation"]:
+            report["cleanup_refused"] = True
+    return report
+
+
+def plan(config):
+    validate(config)
+    return {"prepared_only": True, "not_staged_or_executed": True,
+            "first_delay_seconds": FIRST_DELAY_SECONDS, "retry_delay_seconds": RETRY_DELAY_SECONDS,
+            "accuracy_seconds": 1, "per_attempt_budget_seconds": ATTEMPT_BUDGET_SECONDS,
+            "persistent_failure_retains_recurring_timer": True, "requires_fresh_owned_unit_stem": True,
+            "stop_requires_original_uuid_and_connected_state": True,
+            "original_profile_settings_are_never_modified": True,
+            "cannot_retroactively_repair_offline_host": True, "unit_files": render_units(config)}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("mode", choices=("plan", "attempt", "cleanup"), nargs="?", default="plan")
+    parser.add_argument("--config", required=True)
+    parser.add_argument("--allow-live-host-recovery", action="store_true")
+    args = parser.parse_args()
+    config_path = Path(args.config)
+    config = validate(json.loads(config_path.read_text()))
+    if args.mode == "plan":
+        print(json.dumps(plan(config), indent=2))
+        return
+    require(args.allow_live_host_recovery, "explicit_live_flag_required")
+    require(str(config_path) == config["config_path"], "live_config_path_mismatch")
+    require(str(Path(__file__)) == config["helper_path"], "live_helper_path_mismatch")
+    backend = LiveBackend(config)
+    if args.mode == "cleanup":
+        result = explicit_owned_cleanup(config, backend)
+    else:
+        with owned_lock(config["guard_dir"] + "/attempt.lock", expected_uid=0) as acquired:
+            if not acquired:
+                result = {"schema": 1, "overlapping_attempt_skipped": True, "recurring_retry_retained": True}
+            else:
+                result = attempt(config, backend)
+                destination = Path(config["guard_dir"]) / "attempt-last-safe.json"
+                temporary = destination.with_suffix(".json.tmp")
+                temporary.write_text(json.dumps(result, indent=2) + "\n")
+                os.replace(temporary, destination)
+    print(json.dumps(result))
+
+
+if __name__ == "__main__":
+    main()