Add the agent heartbeat; retune listening so it stops flickering

The face now treats /hermes/state as the agent's heartbeat: when nothing has been
heard for agent_timeout_s (10 s) it logs "agent gone" and sleeps, so "sleeping"
really means "no agent connected". It also logs every state change, which is how
a missed wake message was found: a one-off publish before discovery matches is
lost under volatile QoS.

The live log showed the face flickering between listening and idle about twice a
second, because the detector let go after 0.4 s. Hold is now 1.2 s so pauses
between words do not drop it, and the margin 7 dB over the adaptive floor (quiet
room noise peaks sit about 3 dB over it), with tests for both.

Confirmed by the owner on the real display: gaze directions, and a mouth that
follows their live voice proportionally.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AFN8nRRCiDiMN7ChBpHJhs

CONTEXT.md

@@ -145,7 +145,15 @@ printed enclosure, speakers. Work until those arrive is limited to what the Pi,
   glance is smoothed, held 2.5 s, then drifts back to centre. Sleeping ignores it. The pan sign in
   the renderer was wrong (carried over from the prototype) and is fixed: positive pan = the robot's
   left = the viewer's right, positive tilt = looks up, as the MCP `turn_head` tool documents.
-  Not yet confirmed by eye on the real display.
+  **Confirmed by the owner on the real display (2026-10-04):** the glance directions look right.
+  An earlier run showed nothing because the one-off wake message was lost (volatile QoS drops a
+  sample published before discovery matches); test publishers must republish, as `hermes_mcp`
+  will. The node logs every state change (`state: idle (agent=..., audio=...)`), so behaviour can be
+  checked from `journalctl -u robot-face` without watching the panel.
+- **Agent heartbeat:** `/hermes/state` doubles as the agent's heartbeat. `hermes_mcp` must republish
+  its state about once a second; after `agent_timeout_s` (default 10, 0 disables) with none, the
+  face logs `agent gone, sleeping` and sleeps, so "sleeping" means "no agent connected". Verified:
+  woke on messages, slept 10 s after the last.
 - Not yet: audio levels for the mouth (`/audio/levels`), `hermes_mcp` publishing states, launch
   file in `hermes_bringup`, head/lowpoly shapes, DotStar LEDs, button mute.
 - Open: IPS black level. The backlight is on at full brightness, so black reads dark gray on
@@ -169,11 +177,15 @@ printed enclosure, speakers. Work until those arrive is limited to what the Pi,
   scale is cubic: 63% is -12 dB from the +30 dB maximum, so ALSA `Capture` lands at 46, about +17
   dB). WirePlumber remembers it; recheck after a reboot. The quiet-room floor at this gain is about
   -43 dBFS (preamp and mic noise).
+- **Confirmed by the owner:** the mouth draws their live voice and its size follows how close and
+  loud they are (proportional). The activity detector was retuned after the log showed the face
+  flickering between listening and idle about twice a second: margin 7 dB (quiet-room noise peaks
+  sit at about +3 dB over the floor), hold 1.2 s so pauses between words do not drop it. The
+  flicker fix is covered by unit tests but not yet re-confirmed live (the retest window was silent).
 - Verified on the Pi: the `speaking` path (a tone played to the sink, with no speaker attached, shows
   on the monitor at -13.5 dBFS, `speaking` for its length, mic ignored); levels and scope publish;
-  47 face tests, 8 audio tests. **Not yet verified:** the microphone-to-mouth path with a real voice
+  49 face tests, 10 audio tests. **Not yet verified:** the audio node as a service (it was run as a
-  (one live attempt saw no speech: the cue and the talking did not overlap), and the audio node as a
+  transient unit).
-  service (it was run as a transient unit).
 
 ## Face recognition (opt-in)
 

ros/hermes_audio/hermes_audio/audio_state_node.py

@@ -26,7 +26,7 @@ class AudioStateNode(Node):
         self.n_scope = int(min(2048, max(64, int(param('scope_samples', 256)))))
         self.scope_floor_db = float(param('scope_floor_db', -70.0))
         self.activity = Activity(spk_db=float(param('speaker_threshold_db', -55.0)),
-                                 margin_db=float(param('mic_margin_db', 9.0)))
+                                 margin_db=float(param('mic_margin_db', 7.0)))
         self.levels_pub = self.create_publisher(Float32MultiArray, '/audio/levels', 5)
         self.activity_pub = self.create_publisher(String, '/audio/activity', 5)
         self.scope_pub = {'mic': self.create_publisher(Float32MultiArray, '/audio/scope/mic', 2),

ros/hermes_audio/hermes_audio/levels.py

@@ -34,9 +34,10 @@ class Activity:
     """Decides 'speaking', 'listening' or 'quiet' from the speaker monitor and microphone levels.
 
     speaking:  the speaker monitor is above spk_db for on_s (so a click is not speech); it lets go
-               after hold_s of quiet.
+               after speak_hold_s of quiet.
     listening: the microphone is above the adaptive threshold while not speaking, for on_s; it lets
-               go after hold_s. The threshold is margin_db above a noise floor that drops quickly
+               go after hold_s, long enough (1.2 s) to bridge the pauses between words, so a sentence
+               reads as one listening stretch, not a flicker. The threshold is margin_db above a noise floor that drops quickly
                to new lows and rises slowly (so a long conversation does not become the floor).
     For its first calibrate_s the floor follows the microphone quickly in both directions (a room
     that is already noisy must not read as "listening"), and anything that stays 'listening' for
@@ -44,8 +45,8 @@ class Activity:
     The microphone is ignored while speaking: it would hear the robot's own voice.
     """
 
-    def __init__(self, spk_db: float = -55.0, margin_db: float = 9.0, min_mic_db: float = -62.0,
+    def __init__(self, spk_db: float = -55.0, margin_db: float = 7.0, min_mic_db: float = -62.0,
-                 on_s: float = 0.10, speak_on_s: float = 0.08, hold_s: float = 0.40, speak_hold_s: float = 0.25,
+                 on_s: float = 0.08, speak_on_s: float = 0.08, hold_s: float = 1.2, speak_hold_s: float = 0.25,
                  calibrate_s: float = 3.0, ambient_s: float = 8.0):
         self.spk_db, self.margin_db, self.min_mic_db = spk_db, margin_db, min_mic_db
         self.on_s, self.speak_on_s, self.hold_s, self.speak_hold_s = on_s, speak_on_s, hold_s, speak_hold_s

ros/hermes_audio/test/test_levels.py

@@ -42,9 +42,9 @@ def test_listens_to_a_voice_and_lets_go_after_the_hold():
     assert a.state == 'quiet'
     t = run_activity(a, -35, -100, 0.5, t)
     assert a.state == 'listening'
-    run_activity(a, -62, -100, 0.2, t)
+    run_activity(a, -62, -100, 0.8, t)
-    assert a.state == 'listening'                      # still holding
+    assert a.state == 'listening'                      # still holding, through a pause between words
-    run_activity(a, -62, -100, 1.0, t + 0.2)
+    run_activity(a, -62, -100, 1.0, t + 0.8)
     assert a.state == 'quiet'
 
 
@@ -74,10 +74,32 @@ def test_a_room_that_is_already_noisy_calibrates_instead_of_listening_forever():
     assert a.state == 'quiet' and a.mic_threshold_db > -40
     t = run_activity(a, -25, -100, 0.5, 5.0)           # a voice 18 dB above it
     assert a.state == 'listening'
-    run_activity(a, -43, -100, 1.0, t)
+    run_activity(a, -43, -100, 2.0, t)
     assert a.state == 'quiet'
 
 
+def test_pauses_between_words_do_not_flicker():
+    a = Activity()
+    t = run_activity(a, -43, -100, 5)
+    states = []
+    for _ in range(6):                                 # six words, 0.3 s each with 0.3 s gaps
+        for level, secs in ((-25, 0.3), (-43, 0.3)):
+            for _ in range(int(secs / 0.05)):
+                t += 0.05
+                states.append(a.update(level, -100, t, 0.05))
+    assert states[8:].count('quiet') == 0              # once it starts listening it stays there
+
+
+def test_quiet_room_noise_peaks_stay_below_the_threshold():
+    a = Activity()
+    t = run_activity(a, -43, -100, 5)
+    flicker = []
+    for i in range(200):
+        t += 0.05
+        flicker.append(a.update(-43 + (3 if i % 7 == 0 else 0), -100, t, 0.05))   # peaks 3 dB over the floor
+    assert set(flicker) == {'quiet'}
+
+
 def test_a_new_steady_noise_is_absorbed_after_a_while():
     a = Activity()
     t = run_activity(a, -60, -100, 5)                  # quiet room

ros/hermes_face/hermes_face/arbiter.py

@@ -20,3 +20,21 @@ def resolve_state(agent: str, audio: str) -> str:
     if agent == 'thinking':
         return 'thinking'
     return 'idle'
+
+
+class Heartbeat:
+    """Whether the agent side has been heard from recently. The agent (hermes_mcp) republishes its
+    state about once a second; when that stops, nobody is connected and the face should sleep.
+    timeout_s <= 0 disables the check (always alive)."""
+
+    def __init__(self, timeout_s: float):
+        self.timeout_s = timeout_s
+        self._last = None
+
+    def beat(self, now: float) -> None:
+        self._last = now
+
+    def alive(self, now: float) -> bool:
+        if self.timeout_s <= 0:
+            return True
+        return self._last is not None and now - self._last <= self.timeout_s

ros/hermes_face/hermes_face/face_node.py

@@ -7,6 +7,9 @@ Subscribes:  /hermes/state (std_msgs/String): sleeping idle listening thinking l
              /camera/camera_info (sensor_msgs/CameraInfo): the image size for that, if published
              /audio/activity (std_msgs/String): speaking | listening | quiet, from hermes_audio
              /audio/scope/{mic,speaker} (std_msgs/Float32MultiArray): the real waveform for the mouth
+/hermes/state doubles as the agent's heartbeat: hermes_mcp republishes it about once a second, and
+when it has not been heard for `agent_timeout_s` (default 10, 0 disables) the face goes to sleep, so
+"sleeping" really means "no agent connected".
 Which state shows is decided by hermes_face.arbiter: sleeping (agent only) > speaking > listening >
 thinking > idle; room sound never wakes a sleeping face.
 The face starts in `initial_state` (default `sleeping`), so it runs with nothing else up; whichever
@@ -21,7 +24,7 @@ from sensor_msgs.msg import CameraInfo, JointState
 from std_msgs.msg import Float32MultiArray, String
 from vision_msgs.msg import Detection2DArray
 
-from .arbiter import resolve_state
+from .arbiter import Heartbeat, resolve_state
 from .gaze import Target, gaze_from_pixel, pick_target
 from .render import STATES, FaceRenderer
 
@@ -37,6 +40,8 @@ class FaceNode(Node):
         initial = param('initial_state', 'sleeping')
         self.agent_state = initial if initial in STATES else 'sleeping'
         self.audio_activity = 'quiet'
+        self._logged_state = ''
+        self.heartbeat = Heartbeat(float(param('agent_timeout_s', 10.0)))
         self.audio_drive = bool(param('audio_drive', True))
         self.follow = bool(param('gaze_follow', True))
         self.image_size = (float(param('camera_width', 1296)), float(param('camera_height', 972)))
@@ -69,12 +74,14 @@ class FaceNode(Node):
         self._frames, self._stat_t = 0, self._last
         self.create_timer(1.0 / self.fps, self._tick)
         self.get_logger().info(f'face up: {output}, {self.fps:g} fps, state {self.renderer.state}')
+        self._resolve(time.monotonic())
 
     def _on_state(self, msg: String) -> None:
         if msg.data not in STATES:
             self.get_logger().warn(f'ignoring unknown state {msg.data!r}; known: {", ".join(STATES)}')
             return
         now = time.monotonic()
+        self.heartbeat.beat(now)
         if msg.data == 'looking':          # a timed event: the renderer squints, flashes, then goes back
             self.renderer.set_state('looking', now)
             return
@@ -86,8 +93,14 @@ class FaceNode(Node):
         self._resolve(time.monotonic())
 
     def _resolve(self, now: float) -> None:
+        if self.agent_state != 'sleeping' and not self.heartbeat.alive(now):
+            self.get_logger().info(f'no /hermes/state for {self.heartbeat.timeout_s:g} s: agent gone, sleeping')
+            self.agent_state = 'sleeping'
         if self.renderer.state != 'looking':
             self.renderer.set_state(resolve_state(self.agent_state, self.audio_activity if self.audio_drive else 'quiet'), now)
+        if self.renderer.state != self._logged_state:
+            self._logged_state = self.renderer.state
+            self.get_logger().info(f'state: {self._logged_state} (agent={self.agent_state}, audio={self.audio_activity})')
 
     def _on_joints(self, msg: JointState) -> None:
         pos = dict(zip(msg.name, msg.position))

ros/hermes_face/test/test_arbiter.py

@@ -1,6 +1,6 @@
 import pytest
 
-from hermes_face.arbiter import resolve_state
+from hermes_face.arbiter import Heartbeat, resolve_state
 
 
 @pytest.mark.parametrize('agent,audio,expected', [
@@ -18,3 +18,17 @@ from hermes_face.arbiter import resolve_state
 ])
 def test_priority(agent, audio, expected):
     assert resolve_state(agent, audio) == expected
+
+
+def test_heartbeat_expires_without_beats_and_recovers():
+    hb = Heartbeat(10.0)
+    assert not hb.alive(0.0)                 # never heard from: not alive
+    hb.beat(5.0)
+    assert hb.alive(5.0) and hb.alive(15.0)
+    assert not hb.alive(15.1)
+    hb.beat(20.0)
+    assert hb.alive(25.0)
+
+
+def test_heartbeat_disabled_is_always_alive():
+    assert Heartbeat(0).alive(1e9) and Heartbeat(-1).alive(0.0)