Phase 4: wire sink reactivation into the alarm (finishes item 6)

_reactivate.py shipped in phase 2a with tests but nothing called it, so the
HDMI sink fix was never actually in the alarm's path. WakeAlarm.__init__ now
starts the retry next to _activate_all_outputs(), stopped by the same
_stop_beep event the beep loop watches.

Verified on the host, not just in tests. Starting state was the exact bug:

    alsa_card.pci-0000_01_00.1  Active Profile: off
    sinks: alsa_output.pci-0000_0d_00.6.analog-stereo

After the retry ran, both passes logged:

    2 sink(s) present: alsa_output.pci-0000_01_00.1.hdmi-stereo,
                       alsa_output.pci-0000_0d_00.6.analog-stereo
    default=alsa_output.pci-0000_01_00.1.hdmi-stereo

and the captured card profiles restored cleanly afterwards.

Two wiring details worth keeping:

- _activate_outputs_only() drops what _activate_all_outputs() returns. Only
  the first call sees the user's real audio state; every retry captures the
  alarm's own. Returning one would let a later caller restore the alarm's
  settings as if they were the user's.
- _list_real_sinks_now() re-resolves pactl each pass and reports no sinks
  when it is absent. The retry outlives the startup lookup and runs on a
  thread nobody is watching, where an exception would vanish silently.

_alarm.py was exactly at the 250-line cap, so the flash timers move to
_alarm_flash.py as a mixin, mirroring AlarmTimersMixin. AlarmTimersMixin's
abstract _start_flash_countdown stub had to go: it is first in the MRO and
would shadow the real implementation. main() stays in _alarm.py --
morning_routine/_orchestrator.py:39 hardcodes that path.

conftest stubs the reactivation autouse; without it every WakeAlarm test
would reshape the host's audio routing.

409 tests pass, 100% branch coverage.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EckWmyfetYq3Mi8MAnUudh

wake_alarm/_alarm.py

@@ -1,7 +1,7 @@
 """Wake alarm: escalating beeps behind a fullscreen dismiss challenge.
 
 Started by ``morning-routine.service``, never directly (AGENTS.md). Gate:
-``_gating``. Sound: ``_beep_loop``. Timers: ``_alarm_timers``.
+``_gating``. Sound: ``_beep_loop``. Timers: ``_alarm_timers`` and ``_alarm_flash``.
 """
 
 from __future__ import annotations
@@ -20,6 +20,7 @@ from gatelock import (
 )
 
 from wake_alarm._alarm_display import _restore_display
+from wake_alarm._alarm_flash import AlarmFlashMixin
 from wake_alarm._alarm_state import _AlarmHardware, _AlarmProgress
 from wake_alarm._alarm_timers import AlarmTimersMixin
 from wake_alarm._alarm_view import _build_alarm_view, _ViewGroup
@@ -27,23 +28,17 @@ from wake_alarm._beep_loop import start_beep_thread
 from wake_alarm._challenges import (
     _make_challenge,
 )
-from wake_alarm._constants import (
-    DISMISS_FLASH_SECONDS,
-    DISMISS_ROUNDS_REQUIRED,
-)
+from wake_alarm._constants import DISMISS_ROUNDS_REQUIRED
 from wake_alarm._escape_flow import EscapeHatch, build_tracker
 from wake_alarm._gating import parse_args, should_run_alarm
 from wake_alarm._lock_setup import build_config, claim_arbiter
 from wake_alarm._logging_setup import configure_logging
+from wake_alarm._reactivate import start_alarm_reactivation
 from wake_alarm._rounds import (
     BANNER_LINGER_MS,
-    FLASH_PERIOD_MS,
-    apply_flash_colour,
-    hide_flash_code,
     normalise,
     show_challenge,
     show_dismissal,
-    show_flash_countdown,
     show_round_progress,
     show_wrong_answer,
 )
@@ -60,7 +55,7 @@ from wake_alarm._state import (
 _logger = logging.getLogger(__name__)
 
 
-class WakeAlarm(AlarmTimersMixin):
+class WakeAlarm(AlarmTimersMixin, AlarmFlashMixin):
     """Fullscreen wake alarm with escalating beep and dismiss challenge."""
 
     def __init__(self, *, demo_mode: bool = False) -> None:
@@ -97,6 +92,12 @@ class WakeAlarm(AlarmTimersMixin):
         self._schedule_skip_window_close()
         start_beep_thread(self._stop_beep, self._progress.started_at)
         self._hardware = _AlarmHardware(audio_restore=_activate_all_outputs())
+        # The first activation above runs before the monitor has finished its
+        # HDMI handshake, so the sink that matters most may not exist yet.
+        # Keep retrying in the background until the alarm is dismissed -- only
+        # a fresh set-card-profile can bring that sink into existence, so no
+        # amount of re-enumerating sinks in the beep loop would find it.
+        start_alarm_reactivation(self._stop_beep)
         self._flash_step()
         self._lock.grab_input()
 
@@ -176,22 +177,6 @@ class WakeAlarm(AlarmTimersMixin):
         if self._progress.current_challenge.kind == "flash":
             self._start_flash_countdown()
 
-    def _start_flash_countdown(self) -> None:
-        """Begin the flash countdown: show code then hide it."""
-        self._progress.flash_remaining = DISMISS_FLASH_SECONDS
-        self._flash_tick()
-
-    def _flash_tick(self) -> None:
-        """Decrement flash countdown; replace the displayed code with placeholders."""
-        if not self._active:
-            return
-        if self._progress.flash_remaining > 0:
-            show_flash_countdown(self._view, self._progress.flash_remaining)
-            self._progress.flash_remaining -= 1
-            self.root.after(1000, self._flash_tick)
-        else:
-            hide_flash_code(self._view, self._progress.current_challenge)
-
     def _dismiss_alarm(self, *, earned_skip: bool) -> None:
         """Dismiss the alarm and save state."""
         self._active = False
@@ -211,16 +196,6 @@ class WakeAlarm(AlarmTimersMixin):
         _logger.warning("alarm dismissed via the escape hatch, no skip earned")
         self._dismiss_alarm(earned_skip=False)
 
-    def _flash_step(self) -> None:
-        """Alternate the background colour to keep the screen attention-grabbing."""
-        if not self._active:
-            return
-        cfg = self._view.config
-        colour = cfg.danger if self._progress.flash_on else cfg.bg
-        apply_flash_colour(self.root, self._view.containers, colour)
-        self._progress.flash_on = not self._progress.flash_on
-        self.root.after(FLASH_PERIOD_MS, self._flash_step)
-
     def run(self) -> None:
         """Start the alarm main loop, restoring hardware on every exit path."""
         self._lock.run()

wake_alarm/_alarm_flash.py

@@ -0,0 +1,71 @@
+"""Screen-flash timers: the flash challenge countdown and the colour pulse.
+
+Split out of ``_alarm.py`` for the 250-line cap, as a mixin rather than free
+functions because each of these reschedules itself through ``root.after`` and
+needs the alarm's own view and progress state -- the same shape as
+``AlarmTimersMixin``.
+
+Two independent flashes live here and must not be confused:
+
+* the **challenge** countdown (:meth:`_start_flash_countdown`) shows a code for
+  ``DISMISS_FLASH_SECONDS`` and then hides it, so the user has to memorise it;
+* the **background** pulse (:meth:`_flash_step`) alternates the window colour
+  forever, so the screen keeps demanding attention.
+
+Both stop the moment ``_active`` goes false, which is what keeps a dismissed
+alarm from repainting a window that is on its way out.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from wake_alarm._constants import DISMISS_FLASH_SECONDS
+from wake_alarm._rounds import (
+    FLASH_PERIOD_MS,
+    apply_flash_colour,
+    hide_flash_code,
+    show_flash_countdown,
+)
+
+if TYPE_CHECKING:  # pragma: no cover
+    import tkinter as tk
+
+    from wake_alarm._alarm_state import _AlarmProgress
+    from wake_alarm._alarm_view import _ViewGroup
+
+
+class AlarmFlashMixin:
+    """Self-rescheduling flash timers for the code countdown and background."""
+
+    # Provided by WakeAlarm; declared so type checking sees them.
+    root: tk.Misc
+    _active: bool
+    _progress: _AlarmProgress
+    _view: _ViewGroup
+
+    def _start_flash_countdown(self) -> None:
+        """Begin the flash countdown: show the code, then hide it."""
+        self._progress.flash_remaining = DISMISS_FLASH_SECONDS
+        self._flash_tick()
+
+    def _flash_tick(self) -> None:
+        """Count the code down, then replace it with placeholders."""
+        if not self._active:
+            return
+        if self._progress.flash_remaining > 0:
+            show_flash_countdown(self._view, self._progress.flash_remaining)
+            self._progress.flash_remaining -= 1
+            self.root.after(1000, self._flash_tick)
+        else:
+            hide_flash_code(self._view, self._progress.current_challenge)
+
+    def _flash_step(self) -> None:
+        """Alternate the background colour to keep the screen attention-grabbing."""
+        if not self._active:
+            return
+        cfg = self._view.config
+        colour = cfg.danger if self._progress.flash_on else cfg.bg
+        apply_flash_colour(self.root, self._view.containers, colour)
+        self._progress.flash_on = not self._progress.flash_on
+        self.root.after(FLASH_PERIOD_MS, self._flash_step)

wake_alarm/_alarm_timers.py

@@ -28,6 +28,7 @@ from wake_alarm._rounds import (
 )
 
 if TYPE_CHECKING:  # pragma: no cover
+    from collections.abc import Callable
     import tkinter as tk
 
     from wake_alarm._alarm_state import _AlarmProgress
@@ -48,9 +49,10 @@ class AlarmTimersMixin:
     _view: _ViewGroup
     _escape: EscapeHatch
 
-    def _start_flash_countdown(self) -> None:
-        """Restart the flash countdown; implemented by ``WakeAlarm``."""
-        raise NotImplementedError  # pragma: no cover
+    # Supplied by AlarmFlashMixin, which WakeAlarm also inherits. Declared
+    # rather than stubbed: an abstract stub here would win the MRO (this mixin
+    # is listed first) and shadow the real implementation.
+    _start_flash_countdown: Callable[[], None]
 
     def _schedule_code_refresh(self) -> None:
         """Replace the current challenge periodically.

wake_alarm/_reactivate.py

@@ -25,8 +25,10 @@ from __future__ import annotations
 from collections.abc import Callable, Iterable
 import logging
 import threading
+import time
 
 from wake_alarm._constants import REACTIVATE_ATTEMPT_DELAYS
+from wake_alarm._sinks import _activate_all_outputs, _list_real_sinks, _pactl_path
 
 _Activate = Callable[[], object]
 _ListSinks = Callable[[], Iterable[str]]
@@ -128,3 +130,51 @@ def start_background_reactivation(
     )
     thread.start()
     return thread
+
+
+def _activate_outputs_only() -> None:
+    """Re-run card-profile activation, discarding the captured prior state.
+
+    The return value of :func:`_activate_all_outputs` is the audio state as it
+    was *before* that call, and only the very first call sees the user's real
+    settings -- every retry pass captures the alarm's own already-activated
+    state. Restoring one of those would restore the alarm's settings as if they
+    were the user's, so the retries deliberately return nothing at all rather
+    than handing back a value someone could later wire into ``on_close()``.
+    """
+    _activate_all_outputs()
+
+
+def _list_real_sinks_now() -> list[str]:
+    """Return the currently visible real sinks, or none when pactl is absent.
+
+    :func:`wake_alarm._sinks._list_real_sinks` needs the pactl path, which is
+    resolved fresh each pass: the retry outlives the startup lookup, and a
+    machine without pactl must report "no sinks" rather than raise on a
+    background thread nobody is watching.
+    """
+    pactl = _pactl_path()
+    if pactl is None:
+        return []
+    return _list_real_sinks(pactl)
+
+
+def start_alarm_reactivation(stop: threading.Event) -> threading.Thread:
+    """Start the real reactivation retry for a ringing alarm.
+
+    The wiring point: everything above takes injected callables so it can be
+    tested without an audio stack, and this is the one place that binds them to
+    the real ``pactl``.
+
+    Args:
+        stop: The alarm's beep-stop event, set once it is dismissed.
+
+    Returns:
+        The started daemon thread.
+    """
+    return start_background_reactivation(
+        _activate_outputs_only,
+        _list_real_sinks_now,
+        sleep=time.sleep,
+        stop=stop,
+    )

wake_alarm/tests/conftest.py

@@ -10,6 +10,11 @@ output. Two consequences for tests, both of which bite silently:
   switching on the machine running the tests, and reaches the real
   ``$XDG_RUNTIME_DIR``, where it could stand a *production* locker down.
 
+Sink reactivation adds a fourth: ``WakeAlarm.__init__`` starts a background
+thread that really shells out to ``pactl set-card-profile`` and would reshape
+the *host's* audio routing mid-test. It is stubbed here and re-patched
+explicitly by the tests that assert the wiring.
+
 The escape hatch adds a third: ``WakeAlarm.__init__`` builds its tracker over
 the real ``ESCAPE_HISTORY_FILE``, so an unpatched run both reads a *production*
 budget -- making "the button appears" tests fail on any day the real budget is
@@ -87,6 +92,7 @@ def _hermetic(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Generator[None
             "wake_alarm._escape_flow.ESCAPE_HISTORY_FILE",
             tmp_path / "escape_history.json",
         ),
+        patch("wake_alarm._alarm.start_alarm_reactivation"),
         patch("gatelock._surfaces.tk.Toplevel", side_effect=make_toplevel),
         patch("gatelock._outputs.RandrBackend.create", return_value=None),
         patch("gatelock._outputs.scan_xrandr", return_value=FAKE_OUTPUTS),

wake_alarm/tests/test_alarm_part2.py

@@ -109,6 +109,25 @@ class TestWakeAlarmInit:
         assert alarm._view._views, "a surface must exist per live output"
         alarm._stop_beep.set()  # Stop beep thread
 
+    def test_init_starts_sink_reactivation(
+        self,
+        mock_tk_module: MagicMock,
+    ) -> None:
+        """The retry runs, and stops when the alarm does.
+
+        Without this the HDMI sink stays absent for the whole alarm: its card
+        profile is ``off`` at wake time, and only a fresh set-card-profile can
+        bring the sink into existence -- re-enumerating sinks never finds it.
+        The conftest stub is replaced here so the wiring itself is asserted.
+        """
+        assert mock_tk_module is not None
+        with patch("wake_alarm._alarm.start_alarm_reactivation") as start:
+            alarm = WakeAlarm(demo_mode=True)
+        start.assert_called_once_with(alarm._stop_beep)
+        # The same event the beep loop watches, so dismissal ends both.
+        alarm._stop_beep.set()
+        assert start.call_args.args[0].is_set()
+
     def test_production_mode_fullscreen(
         self,
         mock_tk_module: MagicMock,

wake_alarm/tests/test_reactivate.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import threading
+import time
 
 import pytest
 
@@ -134,3 +135,79 @@ def test_background_thread_is_a_daemon() -> None:
     )
     thread.join(timeout=5)
     assert thread.daemon
+
+
+def test_activate_only_discards_the_captured_state(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Retry passes must not hand back a state anyone could restore.
+
+    Only the *first* activation sees the user's real audio settings; every
+    later pass captures the alarm's own. Returning one would let a future
+    caller restore the alarm's settings as if they were the user's.
+    """
+    calls: list[str] = []
+
+    def fake_activate() -> str:
+        calls.append("activated")
+        return "prior-state"
+
+    monkeypatch.setattr(_reactivate, "_activate_all_outputs", fake_activate)
+    _reactivate._activate_outputs_only()
+    assert calls == ["activated"]
+
+
+def test_list_sinks_now_returns_the_real_sinks(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The zero-arg adapter resolves pactl and forwards the path."""
+    monkeypatch.setattr(_reactivate, "_pactl_path", lambda: "/usr/bin/pactl")
+    monkeypatch.setattr(
+        _reactivate,
+        "_list_real_sinks",
+        lambda pactl: [f"sink-via-{pactl}"],
+    )
+    assert _reactivate._list_real_sinks_now() == ["sink-via-/usr/bin/pactl"]
+
+
+def test_list_sinks_now_is_empty_without_pactl(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A machine with no pactl reports no sinks instead of raising.
+
+    The adapter runs on a background thread nobody is watching, so an
+    exception there would vanish silently.
+    """
+    monkeypatch.setattr(_reactivate, "_pactl_path", lambda: None)
+    assert _reactivate._list_real_sinks_now() == []
+
+
+def test_start_alarm_reactivation_binds_the_real_callables(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The one wiring point passes the real pactl adapters and time.sleep."""
+    captured: dict[str, object] = {}
+
+    sentinel = threading.Thread(target=lambda: None)
+
+    def fake_start(
+        activate: object,
+        list_sinks: object,
+        *,
+        sleep: object,
+        stop: object,
+    ) -> threading.Thread:
+        captured.update(
+            activate=activate, list_sinks=list_sinks, sleep=sleep, stop=stop
+        )
+        return sentinel
+
+    monkeypatch.setattr(_reactivate, "start_background_reactivation", fake_start)
+    stop = threading.Event()
+
+    assert _reactivate.start_alarm_reactivation(stop) is sentinel
+
+    assert captured["activate"] is _reactivate._activate_outputs_only
+    assert captured["list_sinks"] is _reactivate._list_real_sinks_now
+    assert captured["sleep"] is time.sleep
+    assert captured["stop"] is stop