fix: retry ordinary 429s and recover dead LSP streams

crates/hi-agent/src/agent/turn/fast_feedback.rs

@@ -140,6 +140,7 @@ pub(crate) async fn run_fast_feedback(
                 .collect::<Vec<_>>();
             let mut errors = Vec::new();
             let mut saw_confirmed = false;
+            let mut saw_transport_failure = false;
             for (path, diag_state) in lsp.diagnostics_batch(&path_bufs).await {
                 match diag_state {
                     hi_lsp::DiagnosticState::ConfirmedClean { .. } => {
@@ -160,6 +161,7 @@ pub(crate) async fn run_fast_feedback(
                         }
                     }
                     hi_lsp::DiagnosticState::Failed { error, .. } => {
+                        saw_transport_failure = true;
                         ui.status(&format!(
                             "fast check · LSP failed for {}: {error}",
                             path_display(runtime.root(), &path)
@@ -176,7 +178,14 @@ pub(crate) async fn run_fast_feedback(
                 // LSP already found compile-level issues — skip cargo this batch.
                 return report;
             }
-            lsp_checked_clean = saw_confirmed;
+            // Transport death (closed stream, poison) is not a clean bill of
+            // health — fall through to cargo check instead of sealing green.
+            if saw_transport_failure && !saw_confirmed {
+                lsp_unavailable = true;
+                ui.status("fast check · LSP unavailable; falling back to cargo check");
+            } else {
+                lsp_checked_clean = saw_confirmed;
+            }
         }
     }
 

crates/hi-agent/src/agent/turn/loop_.rs

@@ -58,7 +58,8 @@ use super::progress::{
 use super::retry::{
     INCOMPLETE_STATUS, MAX_PROVIDER_OVERLOAD_RETRIES, MAX_TRANSIENT_ROUTE_RETRIES,
     ReviewRepairState, TurnRetryState, delay_label, estimate_tool_schema_tokens,
-    output_cap_retry_tokens, provider_overload_retry_delay, transient_route_retry_delay,
+    output_cap_retry_tokens, provider_error_is_backoff_retryable, provider_overload_retry_delay,
+    transient_route_retry_delay,
 };
 
 impl crate::Agent {
@@ -814,16 +815,23 @@ impl crate::Agent {
                     Err(err)
                         if retry_state.provider_overload_retries
                             < MAX_PROVIDER_OVERLOAD_RETRIES
-                            && hi_ai::provider_error_is_temporary_overload(&err) =>
+                            && provider_error_is_backoff_retryable(&err) =>
                     {
                         ui.assistant_end();
                         self.add_error_usage(&err);
                         self.emit_usage(ui);
                         retry_state.provider_overload_retries += 1;
                         let retry = retry_state.provider_overload_retries;
                         let delay = provider_overload_retry_delay(retry, &err);
+                        let reason = if provider_error_kind(&err)
+                            == Some(ProviderErrorKind::RateLimit)
+                        {
+                            "rate limited"
+                        } else {
+                            "request did not complete"
+                        };
                         ui.nudge(&format!(
-                            "request did not complete; retrying {} ({retry}/{MAX_PROVIDER_OVERLOAD_RETRIES})",
+                            "{reason}; retrying {} ({retry}/{MAX_PROVIDER_OVERLOAD_RETRIES})",
                             delay_label(delay)
                         ));
                         if !delay.is_zero() {

crates/hi-agent/src/agent/turn/retry.rs

@@ -14,9 +14,11 @@ use crate::steering::{EvidenceTracker, ReviewRepairMode};
 pub(super) const MAX_TRANSIENT_ROUTE_RETRIES: u32 = 2;
 pub(super) const TRANSIENT_ROUTE_RETRY_DELAYS: [u64; 2] = [2, 5];
 pub(super) const MAX_TRANSIENT_ROUTE_RETRY_DELAY_SECS: u64 = 30;
-pub(super) const MAX_PROVIDER_OVERLOAD_RETRIES: u32 = 4;
-pub(super) const PROVIDER_OVERLOAD_RETRY_DELAYS: [u64; 4] = [5, 15, 30, 60];
-pub(super) const MAX_PROVIDER_OVERLOAD_RETRY_DELAY_SECS: u64 = 90;
+/// Shared budget for ordinary 429s and temporary provider overload/capacity blips.
+/// Exponential schedule so a sticky throttle has room to clear without hammering.
+pub(super) const MAX_PROVIDER_OVERLOAD_RETRIES: u32 = 8;
+pub(super) const PROVIDER_OVERLOAD_RETRY_DELAYS: [u64; 8] = [1, 2, 4, 8, 16, 32, 64, 120];
+pub(super) const MAX_PROVIDER_OVERLOAD_RETRY_DELAY_SECS: u64 = 120;
 pub(super) const MIN_OUTPUT_CAP_RETRY_TOKENS: u32 = 512;
 pub(super) const INCOMPLETE_STATUS: &str = "turn stopped incomplete";
 
@@ -133,9 +135,14 @@ pub(super) fn provider_retry_delay(
         .get(retry.saturating_sub(1) as usize)
         .copied()
         .unwrap_or(*default_delays.last().unwrap_or(&5));
-    let secs = hi_ai::provider_retry_after_seconds(err)
-        .unwrap_or(default)
-        .min(max_delay_secs);
+    // Prefer the provider's Retry-After when it asks us to wait; treat an explicit
+    // 0 as "retry immediately" (common on overload blips / tests). Missing values
+    // fall through to the exponential table.
+    let secs = match hi_ai::provider_retry_after_seconds(err) {
+        Some(0) => 0,
+        Some(secs) => secs.max(default).min(max_delay_secs),
+        None => default.min(max_delay_secs),
+    };
     if secs == 0 {
         return std::time::Duration::ZERO;
     }
@@ -146,6 +153,14 @@ pub(super) fn provider_retry_delay(
     std::time::Duration::from_secs(secs) + std::time::Duration::from_millis(jitter_ms)
 }
 
+/// Rate limits and temporary overload/capacity errors share the extended backoff budget.
+pub(super) fn provider_error_is_backoff_retryable(err: &anyhow::Error) -> bool {
+    matches!(
+        hi_ai::provider_error_kind(err),
+        Some(hi_ai::ProviderErrorKind::RateLimit)
+    ) || hi_ai::provider_error_is_temporary_overload(err)
+}
+
 pub(super) fn delay_label(delay: std::time::Duration) -> String {
     if delay.is_zero() {
         "now".to_string()

crates/hi-agent/src/tests/retry.rs

@@ -724,6 +724,10 @@ async fn temporary_provider_overload_gets_extended_retry_budget() {
     };
     let (mut agent, requests) = scripted_agent(
         vec![
+            overload(),
+            overload(),
+            overload(),
+            overload(),
             overload(),
             overload(),
             overload(),
@@ -735,17 +739,53 @@ async fn temporary_provider_overload_gets_extended_retry_budget() {
 
     agent.run_turn("go", &mut NullUi).await.unwrap();
 
-    assert_eq!(requests.lock().unwrap().len(), 5);
+    assert_eq!(requests.lock().unwrap().len(), 9);
     assert_eq!(agent.messages().last().unwrap().text(), "recovered");
 }
 
 #[tokio::test]
-async fn ordinary_rate_limit_does_not_use_overload_retry_budget() {
+async fn ordinary_rate_limit_retries_with_backoff_budget() {
+    let limited = || {
+        ProviderStep::ErrorMessage(
+            ProviderErrorKind::RateLimit,
+            r#"API error 429 Too Many Requests: {"error":{"message":"quota exceeded","code":"rate_limit"},"retry_after_seconds":0}"#.into(),
+        )
+    };
     let (mut agent, requests) = scripted_agent(
-        vec![ProviderStep::ErrorMessage(
+        vec![
+            limited(),
+            limited(),
+            ProviderStep::Completion(completion(vec![Content::Text("recovered".into())], 5, 3)),
+        ],
+        config(),
+    );
+
+    agent.run_turn("go", &mut NullUi).await.unwrap();
+
+    assert_eq!(requests.lock().unwrap().len(), 3);
+    assert_eq!(agent.messages().last().unwrap().text(), "recovered");
+}
+
+#[tokio::test]
+async fn ordinary_rate_limit_exhausts_backoff_budget() {
+    let limited = || {
+        ProviderStep::ErrorMessage(
             ProviderErrorKind::RateLimit,
-            r#"API error 429 Too Many Requests: {"error":{"message":"quota exceeded","code":"rate_limit"}}"#.into(),
-        )],
+            r#"API error 429 Too Many Requests: {"error":{"message":"too many requests","code":"rate_limit"},"retry_after_seconds":0}"#.into(),
+        )
+    };
+    let (mut agent, requests) = scripted_agent(
+        vec![
+            limited(),
+            limited(),
+            limited(),
+            limited(),
+            limited(),
+            limited(),
+            limited(),
+            limited(),
+            limited(),
+        ],
         config(),
     );
 
@@ -755,7 +795,8 @@ async fn ordinary_rate_limit_does_not_use_overload_retry_budget() {
         hi_ai::provider_error_kind(&err),
         Some(ProviderErrorKind::RateLimit)
     );
-    assert_eq!(requests.lock().unwrap().len(), 1);
+    // initial attempt + 8 retries
+    assert_eq!(requests.lock().unwrap().len(), 9);
 }
 
 #[tokio::test]

crates/hi-lsp/src/client.rs

@@ -41,12 +41,22 @@ enum ReadOutcome {
     /// No data arrived within the budget; the stream position is untouched.
     Idle,
     /// The stream closed (server exited or pipe broke at a frame boundary).
+    /// The client is poisoned — further I/O is useless until respawn.
     Closed,
     /// A read stalled mid-frame — the stream position is unknown, so this
     /// client is unusable and must be respawned (see [`LspClient::is_poisoned`]).
     Poisoned,
 }
 
+/// Outcome of draining server→client notifications after a document sync.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum DrainOutcome {
+    /// Budget expired or the stream stayed idle; client is still usable.
+    Ok,
+    /// stdout closed or desynced; client is poisoned and must be respawned.
+    Dead,
+}
+
 /// One running language server. Owns the child process and its stdio.
 pub struct LspClient {
     child: Mutex<Child>,
@@ -129,20 +139,33 @@ impl LspClient {
         let mut stdout = self.stdout.lock().await;
         match tokio::time::timeout(budget, stdout.fill_buf()).await {
             Err(_) => return ReadOutcome::Idle,
-            Ok(Err(_)) => return ReadOutcome::Closed,
-            Ok(Ok([])) => return ReadOutcome::Closed,
+            Ok(Err(_)) => {
+                self.mark_poisoned();
+                return ReadOutcome::Closed;
+            }
+            Ok(Ok([])) => {
+                self.mark_poisoned();
+                return ReadOutcome::Closed;
+            }
             Ok(Ok(_)) => {}
         }
         match tokio::time::timeout(MESSAGE_GRACE, read_message(&mut stdout)).await {
             Ok(Some(msg)) => ReadOutcome::Message(msg),
-            Ok(None) => ReadOutcome::Closed,
+            Ok(None) => {
+                self.mark_poisoned();
+                ReadOutcome::Closed
+            }
             Err(_) => {
-                self.poisoned.store(true, Ordering::SeqCst);
+                self.mark_poisoned();
                 ReadOutcome::Poisoned
             }
         }
     }
 
+    fn mark_poisoned(&self) {
+        self.poisoned.store(true, Ordering::SeqCst);
+    }
+
     /// Record a `publishDiagnostics` notification into `pushed_diagnostics`.
     fn capture_notification(&self, v: &Value) {
         if let Some((uri, published)) = parse_published_diagnostics(v) {
@@ -197,7 +220,9 @@ impl LspClient {
             }
             match self.read_one(remaining).await {
                 ReadOutcome::Idle => continue, // deadline re-checked above
-                ReadOutcome::Closed => bail!("LSP server closed the stream"),
+                ReadOutcome::Closed => {
+                    bail!("LSP server closed the stream")
+                }
                 ReadOutcome::Poisoned => bail!(
                     "LSP stream lost sync during `{method}`; the server will be restarted on the next query"
                 ),
@@ -231,9 +256,16 @@ impl LspClient {
     }
 
     pub async fn notify(&self, method: &str, params: Value) -> Result<()> {
+        if self.is_poisoned() {
+            bail!("LSP server closed the stream");
+        }
         let body = json!({ "jsonrpc": "2.0", "method": method, "params": params });
         let mut stdin = self.stdin.lock().await;
-        write_message(&mut stdin, &body.to_string()).await?;
+        if let Err(error) = write_message(&mut stdin, &body.to_string()).await {
+            // Broken pipe / write failure means the child is gone or wedged.
+            self.mark_poisoned();
+            return Err(error.into());
+        }
         Ok(())
     }
 
@@ -245,8 +277,12 @@ impl LspClient {
         // Drain pending notifications (the server pushes diagnostics after
         // didOpen). A short budget: most servers publish within a few hundred
         // ms; the diagnostics method does its own longer wait if needed.
-        self.drain_notifications(Duration::from_millis(500)).await;
-        Ok(())
+        // If the server dies mid-drain, fail the sync so the manager can
+        // respawn and retry instead of treating a dead client as synchronized.
+        match self.drain_notifications(Duration::from_millis(500)).await {
+            DrainOutcome::Ok => Ok(()),
+            DrainOutcome::Dead => bail!("LSP server closed the stream"),
+        }
     }
 
     pub async fn did_change(&self, uri: &str, text: &str) -> Result<()> {
@@ -265,24 +301,26 @@ impl LspClient {
         )
         .await?;
         // Short drain — see did_open. The diagnostics method waits longer.
-        self.drain_notifications(Duration::from_millis(500)).await;
-        Ok(())
+        match self.drain_notifications(Duration::from_millis(500)).await {
+            DrainOutcome::Ok => Ok(()),
+            DrainOutcome::Dead => bail!("LSP server closed the stream"),
+        }
     }
 
     /// Read any pending notifications from stdout, capturing
     /// `publishDiagnostics` into `pushed_diagnostics`. Returns after `wait`
-    /// with no data. Holds the `io` lock so it can't race a concurrent
-    /// request's read loop and eat its response; the two-phase `read_one`
-    /// means an expiring budget can't cancel a frame mid-read (which used to
-    /// desync the stream when a large diagnostics payload straddled the
-    /// deadline).
-    pub async fn drain_notifications(&self, wait: Duration) {
+    /// with no data, or [`DrainOutcome::Dead`] if the stream closed/desynced.
+    /// Holds the `io` lock so it can't race a concurrent request's read loop
+    /// and eat its response; the two-phase `read_one` means an expiring budget
+    /// can't cancel a frame mid-read (which used to desync the stream when a
+    /// large diagnostics payload straddled the deadline).
+    pub async fn drain_notifications(&self, wait: Duration) -> DrainOutcome {
         let _io = self.io.lock().await;
         let deadline = Instant::now() + wait;
         loop {
             let remaining = deadline.saturating_duration_since(Instant::now());
             if remaining.is_zero() {
-                return;
+                return DrainOutcome::Ok;
             }
             match self.read_one(remaining).await {
                 ReadOutcome::Message(msg) => {
@@ -293,7 +331,8 @@ impl LspClient {
                         self.capture_notification(&v);
                     }
                 }
-                ReadOutcome::Idle | ReadOutcome::Closed | ReadOutcome::Poisoned => return,
+                ReadOutcome::Idle => {}
+                ReadOutcome::Closed | ReadOutcome::Poisoned => return DrainOutcome::Dead,
             }
         }
     }
@@ -558,4 +597,40 @@ mod tests {
         assert_eq!(published.version, Some(7));
         assert_eq!(published.items.len(), 1);
     }
+
+    #[tokio::test]
+    async fn mark_poisoned_blocks_further_notifies() {
+        let mut child = tokio::process::Command::new("cat")
+            .stdin(std::process::Stdio::piped())
+            .stdout(std::process::Stdio::piped())
+            .stderr(std::process::Stdio::null())
+            .kill_on_drop(true)
+            .spawn()
+            .expect("spawn cat");
+        let stdin = child.stdin.take().expect("stdin");
+        let stdout = child.stdout.take().expect("stdout");
+        let client = LspClient {
+            child: Mutex::new(child),
+            stdin: Mutex::new(stdin),
+            stdout: Mutex::new(BufReader::new(stdout)),
+            io: Mutex::new(()),
+            poisoned: AtomicBool::new(false),
+            next_id: AtomicU64::new(1),
+            versions: StdMutex::new(HashMap::new()),
+            pushed_diagnostics: StdMutex::new(HashMap::new()),
+            capabilities: Value::Null,
+            root: PathBuf::from("/tmp"),
+        };
+        assert!(!client.is_poisoned());
+        client.mark_poisoned();
+        assert!(client.is_poisoned());
+        let err = client
+            .notify("textDocument/didOpen", json!({}))
+            .await
+            .expect_err("poisoned client must refuse notifies");
+        assert!(
+            format!("{err:#}").contains("closed the stream"),
+            "unexpected error: {err:#}"
+        );
+    }
 }