fix(relay): keep realtime voice heartbeat alive during long Hermes runs

The realtime voice agent killed a turn after ~90s of websocket silence
(client idle watchdog). The relay heartbeat stopped the moment
hermes_run_status left {running, waiting_for_confirmation}, so a long or
background Hermes run could starve it and trip the stall. The heartbeat
now continues while session.hermes_task is unfinished, and the spoken
progress repeat is raised 30s->90s and gated on a coarse status change so
tool-message churn no longer re-narrates.

Adds plugin/tests/test_realtime_heartbeat.py (11 cases).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

plugin/relay/realtime_agent/broker.py

@@ -103,7 +103,9 @@ _PLAYBACK_DRAIN_TIMEOUT_SECONDS = 2.5
 _PRE_HERMES_STATUS_LEAD_SECONDS = 0.75
 _HERMES_PROGRESS_INTERVAL_SECONDS = 5.0
 _HERMES_SPOKEN_PROGRESS_AFTER_SECONDS = 15.0
-_HERMES_SPOKEN_PROGRESS_REPEAT_SECONDS = 30.0
+# Calmer cadence: only re-speak the SAME high-level status this far apart, and
+# only when the coarse status actually changed (see _should_repeat_spoken_status).
+_HERMES_SPOKEN_PROGRESS_REPEAT_SECONDS = 90.0
 _RESUME_TTL_SECONDS = 30.0
 # Max time a completed background result waits for the floor to clear before it
 # is spoken anyway (ADR 33 Tier B result delivery).
@@ -2581,7 +2583,13 @@ class RealtimeAgentHandler:
             await asyncio.sleep(_HERMES_PROGRESS_INTERVAL_SECONDS)
             now = time.time()
             status = session.hermes_run_status
-            if status not in {"running", "waiting_for_confirmation"}:
+            # Keep the heartbeat alive while the underlying run is still in
+            # flight, even if `status` transiently drifts off "running" — the
+            # client kills the turn after ~90s of websocket silence, so this is
+            # the one thing keeping a long/background run's socket warm.
+            if not _should_continue_heartbeat(
+                session.hermes_task, status, session_closed=session.closed
+            ):
                 return
             elapsed_seconds = now - started_at
             message, status_key = _hermes_progress_status(session)
@@ -2589,19 +2597,21 @@ class RealtimeAgentHandler:
                 status_key.startswith("progress:")
                 and "drafting a response" in status_key.lower()
             )
+            coarse_key = _coarse_spoken_status_key(status, status_key)
             should_speak = (
                 speakable_progress
                 and elapsed_seconds >= _HERMES_SPOKEN_PROGRESS_AFTER_SECONDS
                 and session.floor.can_speak(FloorMouth.ANDROID_FILLER)
-                and (
-                    session.hermes_last_spoken_progress_key != status_key
-                    or now - session.hermes_last_spoken_progress_at
-                    >= _HERMES_SPOKEN_PROGRESS_REPEAT_SECONDS
+                and _should_repeat_spoken_status(
+                    now,
+                    session.hermes_last_spoken_progress_at,
+                    session.hermes_last_spoken_progress_key,
+                    coarse_key,
                 )
             )
             if should_speak:
                 session.hermes_last_spoken_progress_at = now
-                session.hermes_last_spoken_progress_key = status_key
+                session.hermes_last_spoken_progress_key = coarse_key
             await self._send(
                 ws,
                 session,
@@ -3582,6 +3592,69 @@ def _tool_status_line(tool_name: str | None, *, started: bool) -> str:
     return f"Finished {label}."
 
 
+def _should_continue_heartbeat(
+    task: asyncio.Task[Any] | None,
+    status: str,
+    *,
+    session_closed: bool,
+) -> bool:
+    """Whether the Hermes run-progress heartbeat should keep ticking.
+
+    The heartbeat is the only thing keeping the realtime websocket from going
+    silent during a long/background Hermes run, and the client kills the turn
+    after ~90s of silence. So the heartbeat must NOT self-terminate just because
+    ``hermes_run_status`` momentarily drifts off ``running`` (e.g. a status that
+    briefly reads ``completed``/``idle`` between SSE bursts on a still-running
+    background run). It keeps ticking while the underlying task is alive and the
+    socket is open; it only stops once the task is actually finished/None or the
+    session has closed.
+    """
+    if session_closed:
+        return False
+    if task is not None and not task.done():
+        # The run is still in flight regardless of the transient status label.
+        return True
+    # No live task: fall back to the status. Keep ticking only while a run is
+    # genuinely active/awaiting input; otherwise the heartbeat has nothing to
+    # guard and should stop.
+    return status in {"running", "waiting_for_confirmation"}
+
+
+def _coarse_spoken_status_key(status: str, status_key: str) -> str:
+    """Collapse a fine-grained ``status_key`` to a coarse high-level key.
+
+    The repeat gate should fire on a *meaningful* status change, not on tool
+    *message* churn. ``status_key`` values like ``progress:<message text>`` vary
+    every time a tool emits a new line even though the high-level state ("Hermes
+    is working") is unchanged. Collapsing those to a single ``progress`` bucket
+    means message-only churn no longer re-flags ``should_speak`` on the repeat
+    cadence, while real transitions (entering/leaving a tool, confirmation,
+    drafting) still register.
+    """
+    if status_key.startswith("progress:"):
+        return f"{status}:progress"
+    return f"{status}:{status_key}"
+
+
+def _should_repeat_spoken_status(
+    now: float,
+    last_spoken_at: float,
+    last_coarse_key: str | None,
+    coarse_key: str,
+    *,
+    repeat_after_seconds: float = _HERMES_SPOKEN_PROGRESS_REPEAT_SECONDS,
+) -> bool:
+    """Whether a *repeat* spoken-progress nudge is warranted.
+
+    Returns True when the coarse high-level status changed since the last spoken
+    progress, OR when the same coarse status has persisted past the (now calmer)
+    repeat window. A brand-new status (no prior spoken key) always qualifies.
+    """
+    if last_coarse_key is None or coarse_key != last_coarse_key:
+        return True
+    return (now - last_spoken_at) >= repeat_after_seconds
+
+
 def _hermes_progress_status(session: RealtimeAgentSession) -> tuple[str, str]:
     if session.hermes_run_status == "waiting_for_confirmation":
         return "Waiting for confirmation.", "confirmation"

plugin/tests/test_realtime_heartbeat.py

@@ -0,0 +1,150 @@
+"""Unit tests for the realtime-agent Hermes run-progress heartbeat helpers.
+
+Covers two behaviours that keep a long/background Hermes voice turn alive and
+calm (see broker.py `_send_hermes_run_progress`):
+
+- `_should_continue_heartbeat`: the heartbeat must keep ticking while the
+  underlying run task is still in flight even if `hermes_run_status` transiently
+  drifts off "running" (otherwise the websocket goes silent and the client kills
+  the turn after ~90s). It only stops once the task is finished/None or the
+  session has closed.
+- `_should_repeat_spoken_status` + `_coarse_spoken_status_key`: spoken progress
+  is re-flagged only on a *meaningful* high-level status change, or after the
+  (now calmer) repeat window — tool *message* churn within the same coarse state
+  no longer re-triggers speech.
+"""
+
+from __future__ import annotations
+
+import unittest
+
+from plugin.relay.realtime_agent.broker import (
+    _HERMES_SPOKEN_PROGRESS_REPEAT_SECONDS,
+    _coarse_spoken_status_key,
+    _should_continue_heartbeat,
+    _should_repeat_spoken_status,
+)
+
+
+class _FakeTask:
+    """Minimal stand-in for an asyncio.Task exposing only `.done()`."""
+
+    def __init__(self, done: bool) -> None:
+        self._done = done
+
+    def done(self) -> bool:
+        return self._done
+
+
+class HeartbeatContinuationTest(unittest.TestCase):
+    def test_continues_while_task_running_even_off_status(self) -> None:
+        # The run task is still in flight; status briefly read "completed" between
+        # SSE bursts on a background run. Heartbeat must NOT self-terminate.
+        running = _FakeTask(done=False)
+        for status in ("running", "waiting_for_confirmation", "completed", "idle", "error"):
+            self.assertTrue(
+                _should_continue_heartbeat(running, status, session_closed=False),
+                msg=f"should keep ticking for live task at status={status!r}",
+            )
+
+    def test_stops_when_session_closed_even_with_live_task(self) -> None:
+        running = _FakeTask(done=False)
+        self.assertFalse(
+            _should_continue_heartbeat(running, "running", session_closed=True)
+        )
+
+    def test_no_task_falls_back_to_status(self) -> None:
+        # No live task: keep ticking only while a run is genuinely active.
+        self.assertTrue(_should_continue_heartbeat(None, "running", session_closed=False))
+        self.assertTrue(
+            _should_continue_heartbeat(None, "waiting_for_confirmation", session_closed=False)
+        )
+        for status in ("completed", "idle", "cancelled", "error"):
+            self.assertFalse(
+                _should_continue_heartbeat(None, status, session_closed=False),
+                msg=f"no task + terminal status={status!r} should stop",
+            )
+
+    def test_finished_task_falls_back_to_status(self) -> None:
+        finished = _FakeTask(done=True)
+        # A finished task behaves like no task: status decides.
+        self.assertTrue(
+            _should_continue_heartbeat(finished, "running", session_closed=False)
+        )
+        self.assertFalse(
+            _should_continue_heartbeat(finished, "completed", session_closed=False)
+        )
+
+
+class CoarseStatusKeyTest(unittest.TestCase):
+    def test_progress_messages_collapse_to_one_bucket(self) -> None:
+        # Two different tool *messages* under the same high-level status collapse
+        # to a single coarse key, so message churn doesn't re-trigger speech.
+        a = _coarse_spoken_status_key("running", "progress:Reading file foo.py")
+        b = _coarse_spoken_status_key("running", "progress:Reading file bar.py")
+        self.assertEqual(a, b)
+        self.assertEqual(a, "running:progress")
+
+    def test_distinct_tool_states_keep_distinct_keys(self) -> None:
+        self.assertNotEqual(
+            _coarse_spoken_status_key("running", "tool:search:running"),
+            _coarse_spoken_status_key("running", "tool:search:done"),
+        )
+        self.assertNotEqual(
+            _coarse_spoken_status_key("running", "tool:search:running"),
+            _coarse_spoken_status_key("waiting_for_confirmation", "confirmation"),
+        )
+
+
+class SpokenStatusRepeatGateTest(unittest.TestCase):
+    def test_first_status_always_speaks(self) -> None:
+        self.assertTrue(
+            _should_repeat_spoken_status(
+                now=100.0,
+                last_spoken_at=0.0,
+                last_coarse_key=None,
+                coarse_key="running:tool:search:running",
+            )
+        )
+
+    def test_meaningful_change_speaks_immediately(self) -> None:
+        # Different coarse key -> speak even though almost no time has passed.
+        self.assertTrue(
+            _should_repeat_spoken_status(
+                now=101.0,
+                last_spoken_at=100.0,
+                last_coarse_key="running:tool:search:running",
+                coarse_key="running:tool:search:done",
+            )
+        )
+
+    def test_same_status_message_churn_does_not_respeak_early(self) -> None:
+        # Same coarse key, only a tool message churned, and the repeat window has
+        # NOT elapsed -> stay quiet.
+        self.assertFalse(
+            _should_repeat_spoken_status(
+                now=120.0,
+                last_spoken_at=100.0,  # 20s < 90s repeat window
+                last_coarse_key="running:progress",
+                coarse_key="running:progress",
+            )
+        )
+
+    def test_same_status_respeaks_after_repeat_window(self) -> None:
+        # Same coarse key, but the (calm) repeat window elapsed -> a gentle nudge.
+        self.assertTrue(
+            _should_repeat_spoken_status(
+                now=100.0 + _HERMES_SPOKEN_PROGRESS_REPEAT_SECONDS + 1.0,
+                last_spoken_at=100.0,
+                last_coarse_key="running:progress",
+                coarse_key="running:progress",
+            )
+        )
+
+    def test_repeat_window_is_calm(self) -> None:
+        # Guards the cadence relaxation: the repeat window is at least 90s.
+        self.assertGreaterEqual(_HERMES_SPOKEN_PROGRESS_REPEAT_SECONDS, 90.0)
+
+
+if __name__ == "__main__":
+    unittest.main()