fix(mcp): idle timeout + escalating backoff for speak-SSE monitor

Two reliability gaps in the /events/speak subscriber:

1. resp.chunk().await had no idle timeout. A backend that accepts the
   TCP connection but stops producing frames (deadlocked SSE endpoint,
   zombie process) would block the task forever without reconnecting.
   The pill window would never surface for agent-initiated speech and
   there would be nothing to log. Backend emits a `:ping` heartbeat
   every 15 s, so 45 s without any data is now treated as a dead
   stream — the task errors out and the reconnect loop takes over.

2. Flat 2 s backoff escalates nowhere. Logs fill with reconnect lines
   when the backend is down for minutes, and a backend that accepts +
   immediately closes connections (no data) spins the loop tightly.
   Backoff now escalates 500 ms → 30 s on unproductive rounds and
   resets only when at least one frame arrives (the connection was
   genuinely productive, not just accepted).

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>

tauri/src-tauri/src/speak_monitor.rs

@@ -13,9 +13,19 @@
 //! The pill webview handles the rest (audio playback, then emits
 //! `dictate:hide` back to Rust when the audio element's `ended` fires).
 //!
-//! The task reconnects on any error with a 2 s backoff. There's no fancy
-//! exponential backoff — the backend either dies with the app or comes back
-//! quickly, and constant 2 s polling is cheap.
+//! Reconnect policy: idle-timeout + escalating backoff. The stream is
+//! infinite by design, so a successful round means "we were receiving
+//! frames and then the backend closed the connection" (typically a
+//! server restart) — reset backoff and reconnect quickly. A failure or
+//! a round that produced no frames escalates backoff up to a 30 s cap so
+//! long-term outages stop filling stderr with reconnect log lines.
+//!
+//! The idle timeout guards against the worst silent-failure mode: a
+//! backend that accepts the TCP connection but stops producing frames
+//! (deadlocked SSE endpoint, zombie process). Without a timeout the
+//! `chunk().await` blocks forever and the task never notices. The
+//! backend emits a `:ping` comment every 15 s, so 45 s without any data
+//! is a reliable signal the stream is dead.
 
 use std::time::Duration;
 
@@ -23,6 +33,14 @@ use tauri::{AppHandle, Emitter};
 
 use crate::{ensure_dictate_window, SERVER_PORT};
 
+const INITIAL_BACKOFF: Duration = Duration::from_millis(500);
+const MAX_BACKOFF: Duration = Duration::from_secs(30);
+/// Backend emits a `:ping` heartbeat every 15 s. Giving the stream 45 s
+/// of idle budget absorbs one missed heartbeat (slow GC pause, brief
+/// backend stall) without being so long that a truly dead stream blocks
+/// the pill from surfacing for minutes.
+const STREAM_IDLE_TIMEOUT: Duration = Duration::from_secs(45);
+
 pub fn spawn_speak_monitor(app: AppHandle) {
     tauri::async_runtime::spawn(async move {
         run(app).await;
@@ -38,19 +56,45 @@ async fn run(app: AppHandle) {
             return;
         }
     };
+
+    let mut backoff = INITIAL_BACKOFF;
+    let mut attempt: u32 = 0;
+
     loop {
-        if let Err(e) = stream_once(&client, &url, &app).await {
-            eprintln!("speak_monitor: stream err: {e}");
+        let stream_result = stream_once(&client, &url, &app).await;
+        let had_success = matches!(stream_result, Ok(true));
+
+        if had_success {
+            backoff = INITIAL_BACKOFF;
+            attempt = 0;
+        } else {
+            attempt += 1;
+            let reason = match stream_result {
+                Ok(_) => "stream closed without data".to_string(),
+                Err(e) => format!("stream err: {e}"),
+            };
+            eprintln!(
+                "speak_monitor: {reason} (attempt {attempt}, retry in {:?})",
+                backoff
+            );
+        }
+
+        tokio::time::sleep(backoff).await;
+        if !had_success {
+            backoff = (backoff * 2).min(MAX_BACKOFF);
         }
-        tokio::time::sleep(Duration::from_secs(2)).await;
     }
 }
 
+/// Consume the SSE stream until it closes or errors. Returns `Ok(true)`
+/// if at least one frame was received (the connection was genuinely
+/// productive), `Ok(false)` on a clean but empty close, and `Err` for
+/// any connection or parse failure.
 async fn stream_once(
     client: &reqwest::Client,
     url: &str,
     app: &AppHandle,
-) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+) -> Result<bool, Box<dyn std::error::Error + Send + Sync>> {
     let mut resp = client
         .get(url)
         .header("Accept", "text/event-stream")
@@ -60,7 +104,21 @@ async fn stream_once(
         return Err(format!("speak_monitor: backend returned {}", resp.status()).into());
     }
     let mut buf = String::new();
-    while let Some(chunk) = resp.chunk().await? {
+    let mut saw_data = false;
+    loop {
+        let chunk = match tokio::time::timeout(STREAM_IDLE_TIMEOUT, resp.chunk()).await {
+            Ok(Ok(Some(chunk))) => chunk,
+            Ok(Ok(None)) => return Ok(saw_data),
+            Ok(Err(e)) => return Err(Box::new(e)),
+            Err(_) => {
+                return Err(format!(
+                    "no data for {:?} (heartbeat should arrive every 15 s)",
+                    STREAM_IDLE_TIMEOUT
+                )
+                .into())
+            }
+        };
+        saw_data = true;
         buf.push_str(std::str::from_utf8(&chunk)?);
         // sse-starlette emits CRLF framing; the spec also permits LF, so
         // handle either. Drain whichever separator appears first.
@@ -79,7 +137,6 @@ async fn stream_once(
             }
         }
     }
-    Ok(())
 }
 
 /// Parse a single SSE frame into (event_name, data_json).