The Watchdog

# labs/watchdog.py
import time


class Watchdog:
    def __init__(self, name: str, timeout: float = 30.0) -> None:
        self.name = name
        self.timeout = timeout
        self._last_beat = time.monotonic()

    def heartbeat(self) -> None:
        self._last_beat = time.monotonic()

    def silence(self) -> float:
        return time.monotonic() - self._last_beat

    def is_alive(self) -> bool:
        return self.silence() <= self.timeout


if __name__ == "__main__":
    wd = Watchdog("llm", timeout=0.5)
    print(f"fresh beat: silence={wd.silence():.2f}s alive={wd.is_alive()}")
    time.sleep(0.6)
    print(f"after 0.6s: silence={wd.silence():.2f}s alive={wd.is_alive()}")

$ uv run python -m labs.watchdog
fresh beat: silence=0.00s alive=True
after 0.6s: silence=0.60s alive=False

A boolean flag would need somebody to set it to false, and the one component that knows it is stuck is the one component that has stopped executing. Storing the moment of the last heartbeat turns the question into subtraction, and subtraction needs no cooperation from anyone. Note the clock: time.monotonic() counts seconds since an arbitrary point and only ever goes up, so a clock correction landing mid-conversation cannot make the silence negative or push it past the limit. The scheduler in chapter 19 chose that clock for the same reason. Your second figure may read 0.60 or 0.61, since sleep promises a floor and not a ceiling.

import threading
from typing import Callable


class Watchdog:
    def __init__(self, name: str, timeout: float = 30.0,
                 on_timeout: Callable[[str, float], None] | None = None,
                 poll: float = 0.1) -> None:
        self.name = name
        self.timeout = timeout
        self.poll = poll
        self.on_timeout = on_timeout or self._report
        self._last_beat = time.monotonic()
        self._running = False
        self._thread: threading.Thread | None = None

    def _report(self, name: str, silence: float) -> None:
        print(f"[watchdog] {name} silent for {silence:.1f}s -- recovering")

    def _watch(self) -> None:
        while self._running:
            silence = self.silence()
            if silence > self.timeout:
                self.on_timeout(self.name, silence)
                self._last_beat = time.monotonic()   # re-arm: one alarm per stall
            time.sleep(self.poll)

    def start(self) -> None:
        self._running = True
        self._thread = threading.Thread(
            target=self._watch, name=f"watchdog-{self.name}", daemon=True)
        self._thread.start()

    def stop(self) -> None:
        self._running = False


if __name__ == "__main__":
    wd = Watchdog("llm", timeout=0.5)
    wd.start()
    time.sleep(0.75)          # feed it nothing
    wd.stop()
    print("main thread done")

$ uv run python -m labs.watchdog
[watchdog] llm silent for 0.6s -- recovering
main thread done

Three decisions live in that small class. daemon=True means the watch thread never keeps the process alive; when your main thread returns, Python tears the watcher down instead of hanging on exit forever, and a supervisor that prevents shutdown is its own kind of failure. The callback takes the name and the measured silence, so a recovery routine can log which stage went quiet and for how long instead of announcing that something, somewhere, stopped. And the assignment to _last_beat immediately after firing is the load-bearing line of the whole chapter: it re-arms the timer so a single stall produces a single alarm. Remove it and you get the failure box below.

def fake_turn(wd: Watchdog) -> None:
    """Stand-in for capture, transcribe, think, speak."""
    wd.heartbeat()
    time.sleep(0.05)


if __name__ == "__main__":
    stalls: list[tuple[str, float]] = []
    wd = Watchdog("llm", timeout=0.5,
                  on_timeout=lambda name, s: stalls.append((name, s)))
    wd.start()

    for i in range(3):
        fake_turn(wd)
        print(f"turn {i + 1} answered")
        time.sleep(0.2)
    print(f"healthy run: {len(stalls)} stalls recorded")

    time.sleep(1.05)          # the stage wedges; no more heartbeats
    wd.stop()
    print(f"after the stall: {len(stalls)} stalls recorded")
    for name, seconds in stalls:
        print(f"  {name} silent {seconds:.1f}s")

$ uv run python -m labs.watchdog
turn 1 answered
turn 2 answered
turn 3 answered
healthy run: 0 stalls recorded
after the stall: 2 stalls recorded
  llm silent 0.6s
  llm silent 0.5s

Both numbers are predictable, and predicting them is how you learn to size a timeout. During the healthy run a heartbeat lands every 0.25 seconds against a 0.5 second limit, so the silence never reaches the threshold and the count stays at zero. Then the feeding stops with 0.25 seconds already on the clock, and 1.05 more brings the total silence to 1.3. The first alarm fires just past 0.5, the re-arm restarts the window, the second fires just past 1.1, and a third would need 1.7 seconds that the run never gets to. Recording the events in a list instead of printing them makes the behaviour something a test can assert, and timing bugs that only appear under a stall are miserable to reproduce by hand. Your figures will land on 0.5 or 0.6 depending on where the poll falls.

# labs/watchdog.py -- full file
import threading
import time
from typing import Callable


class Watchdog:
    """Fires on_timeout(name, silence) when heartbeats stop arriving."""

    def __init__(self, name: str, timeout: float = 30.0,
                 on_timeout: Callable[[str, float], None] | None = None,
                 poll: float = 0.1) -> None:
        self.name = name
        self.timeout = timeout
        self.poll = poll
        self.on_timeout = on_timeout or self._report
        self._last_beat = time.monotonic()
        self._running = False
        self._thread: threading.Thread | None = None

    def _report(self, name: str, silence: float) -> None:
        print(f"[watchdog] {name} silent for {silence:.1f}s -- recovering")

    def heartbeat(self) -> None:
        self._last_beat = time.monotonic()

    def silence(self) -> float:
        return time.monotonic() - self._last_beat

    def is_alive(self) -> bool:
        return self.silence() <= self.timeout

    def _watch(self) -> None:
        while self._running:
            silence = self.silence()
            if silence > self.timeout:
                self.on_timeout(self.name, silence)
                self._last_beat = time.monotonic()   # re-arm: one alarm per stall
            time.sleep(self.poll)

    def start(self) -> None:
        self._running = True
        self._thread = threading.Thread(
            target=self._watch, name=f"watchdog-{self.name}", daemon=True)
        self._thread.start()

    def stop(self) -> None:
        self._running = False
        if self._thread is not None:
            self._thread.join(timeout=2 * self.poll)
            self._thread = None

stop() is cooperative, and the join says so out loud. Setting _running to false does not interrupt anything; the watch thread notices on its next trip around the loop, which is at most one poll away, and the join waits that long before giving up. A caller who wants the watcher provably finished before closing a file gets it. Nobody who forgets to call stop() is punished, because the daemon flag handles process exit anyway.

# labs/supervised_link.py
import threading
import time

import serial

from labs.hardware_link import PORT, command, open_link
from labs.watchdog import Watchdog


class SupervisedLink:
    """The serial link, plus a watchdog and a reopen path."""

    def __init__(self, port: str = PORT, timeout: float = 5.0) -> None:
        self.port = port
        self.link = open_link(port)
        self.needs_reopen = threading.Event()
        self.watchdog = Watchdog("link", timeout=timeout, on_timeout=self._flag)
        self.watchdog.start()
        print(f"[link] {port} open")

    def _flag(self, name: str, silence: float) -> None:
        print(f"[watchdog] {name} silent for {silence:.1f}s")
        self.needs_reopen.set()

    def ping(self) -> None:
        started = time.perf_counter()
        try:
            reply = command(self.link, {"command": "ping"})
        except (serial.SerialException, ConnectionError) as exc:
            print(f"[warn] link lost: {exc}")
            return
        if reply.get("status") == "ok":
            self.watchdog.heartbeat()
            print(f"ping ok ({(time.perf_counter() - started) * 1000:.0f} ms)")

    def service(self) -> None:
        if not self.needs_reopen.is_set():
            return
        print(f"[link] reopening {self.port}")
        try:
            self.link.close()
            self.link = open_link(self.port)
        except serial.SerialException as exc:
            print(f"[link] reopen failed: {exc}")
            return                      # flag stays set; try again next pass
        self.needs_reopen.clear()
        self.watchdog.heartbeat()


if __name__ == "__main__":
    sup = SupervisedLink()
    for _ in range(12):
        sup.service()
        sup.ping()
        time.sleep(1.0)
    sup.watchdog.stop()

$ uv run python -m labs.supervised_link
[link] /dev/ttyUSB0 open
ping ok (11 ms)
ping ok (10 ms)
ping ok (11 ms)
[warn] link lost: device reports readiness to read but returned no data
[warn] link lost: write failed: [Errno 5] Input/output error
[warn] link lost: write failed: [Errno 5] Input/output error
[warn] link lost: write failed: [Errno 5] Input/output error
[watchdog] link silent for 5.0s
[link] reopening /dev/ttyUSB0
[link] reopen failed: could not open port /dev/ttyUSB0: [Errno 2] No such file or directory
[warn] link lost: write failed: [Errno 5] Input/output error
[link] reopening /dev/ttyUSB0
ping ok (11 ms)
ping ok (11 ms)

Measured on the bench with an ESP32 devkit on /dev/ttyUSB0, the USB cable pulled by hand after the third ping and pushed back in around the tenth second. Your timings, your port name and your kernel's wording for a vanished device will all vary. What should not vary is the sequence: three healthy pings, a run of errors while the board is gone, exactly one watchdog line, a reopen that fails because the device node no longer exists, and a later reopen that works.

The important design choice is that the callback does not touch the serial port; it sets a threading.Event and returns. Recovery runs on the watch thread, and that thread has no idea whether the main thread is halfway through writing a frame, so closing the port underneath it would turn one clean failure into a corrupted one. The watcher raises a flag, the thread that owns the port acts on it in service(), and only the flag crosses between threads. Recovery also has to survive failing: the early return leaves the event set, so a board that takes twenty seconds to come back is retried once per pass with no extra machinery.