MOR-1191: drive the TX safety supervisor so the watchdog can fire

src/rigplane/core/tx_safety.py

@@ -146,6 +146,8 @@ class TxSafetySnapshot:
     release_last_error: str | None
     active_attempt: ProviderAttempt | None
     watchdog_deadline_monotonic: float | None
+    # Configured *and* driven: a watchdog nothing ticks cannot fire, and this
+    # field must never advertise one (MOR-1191).
     watchdog_enabled: bool
     external_conflict: bool
 
@@ -249,6 +251,7 @@ def __init__(
         self._active: ProviderAttempt | None = None
         self._cancel_pending: CancelProviderAttempt | None = None
         self._watchdog_deadline: float | None = None
+        self._driven = False
 
     @property
     def snapshot(self) -> TxSafetySnapshot:
@@ -290,7 +293,7 @@ def snapshot(self) -> TxSafetySnapshot:
             release_last_error=self._release.error if self._release else None,
             active_attempt=self._active,
             watchdog_deadline_monotonic=self._watchdog_deadline,
-            watchdog_enabled=self._watchdog_seconds is not None,
+            watchdog_enabled=self._watchdog_seconds is not None and self._driven,
             external_conflict=managed_on and not confirmed,
         )
 
@@ -470,6 +473,7 @@ def settle_attempt(
         return self._result(TxOutcome.APPLIED)
 
     def tick(self) -> TxTransition:
+        self._driven = True
         now = self._clock()
         effects: tuple[TxEffect, ...] = ()
         if (

src/rigplane/runtime/managed_radio_runtime.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 import time
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, replace
@@ -18,6 +19,8 @@
     TxTransition,
 )
 
+logger = logging.getLogger(__name__)
+
 TxService = Callable[[TxSafetySupervisor, TxTransition], Awaitable[None]]
 ProviderRelease = Callable[[], Awaitable[None]]
 _PttObserver = Callable[[ProviderPttObservation], None]
@@ -62,9 +65,13 @@ def __init__(
         clock: Clock | None = None,
         id_factory: IdFactory | None = None,
         shutdown_timeout_seconds: float = 3.0,
+        tick_interval_seconds: float = 0.25,
     ) -> None:
-        if not 0 < shutdown_timeout_seconds < float("inf"):
-            raise ValueError("shutdown_timeout_seconds must be finite-positive")
+        if not all(
+            0 < seconds < float("inf")
+            for seconds in (shutdown_timeout_seconds, tick_interval_seconds)
+        ):
+            raise ValueError("shutdown timeout and tick interval must be positive")
         self.target_id, self._clock = target_id, clock or time.monotonic
         self._tx_safety = TxSafetySupervisor(clock=self._clock, id_factory=id_factory)
         self._provider_lifecycle, self._provider_generation = provider_lifecycle, 0
@@ -78,6 +85,8 @@ def __init__(
         self._observation_version = 0
         self._shutdown_task: _ShutdownTask | None = None
         self._shutdown_pending, self._shutdown_timeout = False, shutdown_timeout_seconds
+        self._tick_interval = tick_interval_seconds
+        self._tick_task: asyncio.Task[None] | None = None
         self._effect_host = _ManagedTxEffectHost(
             self._clock, self._host_write, self._host_read, self._host_retire
         )
@@ -91,6 +100,31 @@ async def _service_effects(self, transition: TxTransition) -> None:
         if transition.effects:
             await self._service(self._tx_safety, transition)
 
+    async def _tick_loop(self) -> None:
+        """The production driver of ``tick``: max key-down plus the timed retry.
+
+        Only ``_lifecycle_lock`` is taken, and only around the reducer call: it
+        is the lock that guards supervisor mutation, while ``_lifecycle_change``
+        guards provider identity, which a tick never changes. Waiting on the
+        latter would park the watchdog behind a retirement — exactly when a
+        keyed rig most needs it. The loop retires itself once the lease is gone
+        so an idle target costs nothing and leaves no task behind.
+        """
+        try:
+            while True:
+                async with self._lifecycle_lock:
+                    if self._shutdown_pending or self.tx_snapshot.lease_id is None:
+                        self._tick_task = None
+                        return
+                    transition = self._tx_safety.tick()
+                try:
+                    await self._service_effects(transition)
+                except Exception:
+                    logger.warning("managed TX tick service failed", exc_info=True)
+                await asyncio.sleep(self._tick_interval)
+        except asyncio.CancelledError:
+            pass
+
     def _not_ready(self) -> TxTransition:
         return TxTransition(TxOutcome.NOT_READY, self.tx_snapshot)
 
@@ -290,6 +324,8 @@ async def request_on(self, owner: TxOwner) -> TxTransition:
             ):
                 return self._not_ready()
             transition = self._tx_safety.request_on(owner)
+            if self._tick_task is None and transition.snapshot.lease_id is not None:
+                self._tick_task = asyncio.create_task(self._tick_loop())
         await self._service_effects(transition)
         return transition
 
@@ -338,6 +374,10 @@ async def _complete_shutdown(
         self, transition: TxTransition, release_provider: ProviderRelease
     ) -> tuple[TxTransition, BaseException | None]:
         error: BaseException | None = None
+        if (ticker := self._tick_task) is not None:
+            self._tick_task = None
+            ticker.cancel()
+            await asyncio.gather(ticker, return_exceptions=True)
         try:
             if transition.outcome is not TxOutcome.NOOP:
                 await asyncio.wait_for(

tests/test_managed_tx_watchdog_ticker.py

@@ -0,0 +1,152 @@
+"""MOR-1191: the managed TX max-key-down watchdog needs a production driver.
+
+``TxSafetySupervisor.tick`` is the only path to the watchdog and to the timed
+retry of a failed OFF, and nothing under ``src`` ever called it: a rig keyed
+through the managed path stayed keyed with no bound at all. No test here calls
+``tick`` -- the whole defect is that nothing did -- so each one advances a fake
+clock, lets the loop run, and watches the wire.
+
+A real supervisor, the real effect service and a hand-rolled provider drive
+every case: a scripted double would answer whatever it was told, and a
+``MagicMock`` satisfies a ``runtime_checkable`` protocol on 3.11 but not on
+3.12+ (gh-102433). ``_Provider`` is imported rather than copied so both suites
+watch the same wire.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+from collections.abc import Callable
+
+from rigplane.core.tx_safety import (
+    TxOwner,
+    TxPhase,
+    TxReleaseReason,
+    TxSafetySupervisor,
+    TxSource,
+    TxTransition,
+)
+from rigplane.runtime.managed_radio_runtime import ManagedRadioRuntime, TxService
+from rigplane.runtime.managed_tx_effect_service import managed_tx_effect_service
+from test_web_recovery_durable_off import _Provider
+
+_OWNER = TxOwner(TxSource.WEBSOCKET, "ws-1")
+_TICK = 0.002
+
+
+class _Clock:
+    """Monotonic only when the test says so."""
+
+    def __init__(self) -> None:
+        self.now = 1_000.0
+
+    def __call__(self) -> float:
+        return self.now
+
+
+class _Managed:
+    """A managed runtime plus everything the tests need to watch it."""
+
+    def __init__(self) -> None:
+        self.clock, self.log = _Clock(), []
+        self.serviced: list[TxTransition] = []
+        self.provider = _Provider(self.log)
+        self.runtime = ManagedRadioRuntime(
+            "watchdog",
+            service_factory=self._factory,
+            provider_lifecycle=self.provider,
+            clock=self.clock,
+            tick_interval_seconds=_TICK,
+        )
+
+    def _factory(self, host: object) -> TxService:
+        inner = managed_tx_effect_service(host)
+
+        async def service(sup: TxSafetySupervisor, moved: TxTransition) -> None:
+            self.serviced.append(moved)
+            await inner(sup, moved)
+
+        return service
+
+
+async def _armed(*, key: bool = True) -> _Managed:
+    """Bring the provider up, seed the OFF ``request_on`` demands, then key."""
+    managed = _Managed()
+    await managed.runtime.replace_provider(ready=True)
+    await managed.runtime.request_fresh_ptt()
+    if key:
+        assert (await managed.runtime.request_on(_OWNER)).snapshot.lease_id
+    managed.log.clear()
+    managed.serviced.clear()
+    return managed
+
+
+async def _settles(predicate: Callable[[], bool], timeout: float = 2.0) -> None:
+    """Let real tick intervals elapse until the driver has done its work."""
+    deadline = time.monotonic() + timeout
+    while not predicate():
+        assert time.monotonic() < deadline, "the ticker never got there"
+        await asyncio.sleep(_TICK)
+
+
+async def test_a_lease_held_past_max_key_down_is_dekeyed_on_the_wire() -> None:
+    """Acceptance 1: the OFF reaches the provider, with no call from the test."""
+    managed = await _armed()
+    assert managed.runtime.tx_snapshot.phase is TxPhase.KEYED
+
+    managed.clock.now += 181.0
+    await _settles(lambda: managed.runtime.tx_snapshot.phase is TxPhase.IDLE)
+
+    assert managed.log == ["ptt(off)", "read_ptt"]
+    assert len(managed.serviced) == 1  # the effects reached the provider, not a void
+    reason = managed.serviced[0].snapshot.release_reason
+    assert reason is TxReleaseReason.BACKEND_MAX_KEY_DOWN
+    # Nothing left to watch: the driver retires instead of idling forever.
+    await _settles(lambda: managed.runtime._tick_task is None)
+
+
+async def test_a_refused_off_retries_on_its_own_schedule() -> None:
+    """Acceptance 2: the clock brings it back -- no reconnect, no second call."""
+    managed = await _armed()
+    managed.provider.write_failures = 1
+
+    lease = managed.runtime.tx_snapshot.lease_id or ""
+    await managed.runtime.request_off(_OWNER, lease)
+    assert managed.log == ["ptt(off)"]
+    assert managed.runtime.tx_snapshot.phase is TxPhase.FAULTED
+
+    await asyncio.sleep(_TICK * 20)  # many ticks, but the retry is not due yet
+    assert managed.log == ["ptt(off)"]
+
+    managed.clock.now += 0.25  # retry_schedule_seconds[0]
+    await _settles(lambda: managed.runtime.tx_snapshot.phase is TxPhase.IDLE)
+    assert managed.log == ["ptt(off)", "ptt(off)", "read_ptt"]
+
+
+async def test_the_ticker_stops_at_shutdown_and_never_fires_again() -> None:
+    """Acceptance 3: shutdown owns the last word and leaves no live task."""
+    managed = await _armed()
+    ticker = managed.runtime._tick_task
+    assert ticker is not None
+
+    await managed.runtime.shutdown(release_provider=lambda: asyncio.sleep(0))
+
+    assert managed.log == ["ptt(off)", "read_ptt"]
+    assert managed.runtime._tick_task is None and ticker.done()
+    managed.clock.now += 10_000.0
+    await asyncio.sleep(_TICK * 20)
+    assert managed.log == ["ptt(off)", "read_ptt"]
+
+
+async def test_nothing_ticks_until_a_lease_exists_and_the_signal_says_so() -> None:
+    """Acceptance 4 and 5: no lease, no driver, no cost, no watchdog claimed."""
+    managed = await _armed(key=False)
+
+    await asyncio.sleep(_TICK * 20)
+    assert managed.runtime._tick_task is None and managed.log == []
+    assert not managed.runtime.tx_snapshot.watchdog_enabled
+
+    await managed.runtime.request_on(_OWNER)
+    await _settles(lambda: managed.runtime.tx_snapshot.watchdog_enabled)
+    assert managed.runtime._tick_task is not None