diff --git a/crates/hi-agent/src/agent/turn/fast_feedback.rs b/crates/hi-agent/src/agent/turn/fast_feedback.rs index b547deb1..5092966f 100644 --- a/crates/hi-agent/src/agent/turn/fast_feedback.rs +++ b/crates/hi-agent/src/agent/turn/fast_feedback.rs @@ -140,6 +140,7 @@ pub(crate) async fn run_fast_feedback( .collect::>(); 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; + } } } diff --git a/crates/hi-agent/src/agent/turn/loop_.rs b/crates/hi-agent/src/agent/turn/loop_.rs index 29ce36d1..c511e33d 100644 --- a/crates/hi-agent/src/agent/turn/loop_.rs +++ b/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,7 +815,7 @@ 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); @@ -822,8 +823,15 @@ impl crate::Agent { 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() { diff --git a/crates/hi-agent/src/agent/turn/retry.rs b/crates/hi-agent/src/agent/turn/retry.rs index 52ae8559..7e7de3ad 100644 --- a/crates/hi-agent/src/agent/turn/retry.rs +++ b/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() diff --git a/crates/hi-agent/src/tests/retry.rs b/crates/hi-agent/src/tests/retry.rs index 91ff30ff..cc2ac1a0 100644 --- a/crates/hi-agent/src/tests/retry.rs +++ b/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] diff --git a/crates/hi-lsp/src/client.rs b/crates/hi-lsp/src/client.rs index 47b069ea..6e3e5e6d 100644 --- a/crates/hi-lsp/src/client.rs +++ b/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, @@ -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:#}" + ); + } } diff --git a/crates/hi-lsp/src/manager.rs b/crates/hi-lsp/src/manager.rs index b36c26fb..7d17d1b3 100644 --- a/crates/hi-lsp/src/manager.rs +++ b/crates/hi-lsp/src/manager.rs @@ -306,14 +306,40 @@ impl LspManager { let language_id = language_id_for_path(&path).unwrap_or_else(|| lang.language_id()); client.did_open(&uri, language_id, text).await }; - if let Err(e) = result { - // The hash was inserted optimistically above, but the server never - // received the notify — leaving it would make every future sync - // skip this content as "unchanged". - self.synced.lock().unwrap().remove(&uri); - return Err(e); + if result.is_ok() { + return Ok(()); + } + let err = result.err().expect("checked is_ok above"); + // The hash was inserted optimistically above, but the server never + // received the notify — leaving it would make every future sync + // skip this content as "unchanged". + self.synced.lock().unwrap().remove(&uri); + + // Dead/poisoned servers used to fail every subsequent file in the + // batch with the same "closed the stream" noise. Respawn once and + // retry as a fresh didOpen against the replacement. + if !is_recoverable_transport_error(&err) { + return Err(err); + } + self.ensure(lang).await?; + let client = { + let servers = self.servers.lock().await; + servers + .get(&lang) + .with_context(|| format!("no LSP server for {lang:?} after restart"))? + .clone() + }; + // After a restart the replacement has never seen this URI, so force + // didOpen even if we thought the doc was already open on the dead server. + self.synced.lock().unwrap().insert(uri.clone(), hash); + let language_id = language_id_for_path(&path).unwrap_or_else(|| lang.language_id()); + match client.did_open(&uri, language_id, text).await { + Ok(()) => Ok(()), + Err(retry_err) => { + self.synced.lock().unwrap().remove(&uri); + Err(retry_err) + } } - Ok(()) } /// Close a deleted or no-longer-relevant document and discard all cached @@ -386,7 +412,14 @@ impl LspManager { // Give push-only servers a bounded opportunity to publish an explicit // empty/nonempty result for this version. - client.drain_notifications(Duration::from_secs(10)).await; + if client.drain_notifications(Duration::from_secs(10)).await + == crate::client::DrainOutcome::Dead + { + return DiagnosticState::Failed { + document_version: Some(version), + error: "LSP server closed the stream".into(), + }; + } if let Some(pushed) = client.get_pushed_diagnostics(uri) && publication_matches_document(&pushed, version) { @@ -506,13 +539,33 @@ impl LspManager { continue; } let state = match tokio::fs::read_to_string(&path).await { - Ok(text) => match self.sync_document(&path, &text).await { - Ok(()) => self.diagnostic_state(&path).await, - Err(error) => DiagnosticState::Failed { - document_version: None, - error: format!("synchronizing document: {error:#}"), - }, - }, + Ok(text) => { + let first = match self.sync_document(&path, &text).await { + Ok(()) => self.diagnostic_state(&path).await, + Err(error) => DiagnosticState::Failed { + document_version: None, + error: format!("synchronizing document: {error:#}"), + }, + }; + // One more recovery pass for transport death discovered + // after sync (e.g. closed stream during diagnostic drain). + // `sync_document` already respawns on its own notify/drain + // failures; this covers the post-sync diagnostic path. + match &first { + DiagnosticState::Failed { error, .. } + if is_recoverable_transport_error(&anyhow::anyhow!("{error}")) => + { + match self.sync_document(&path, &text).await { + Ok(()) => self.diagnostic_state(&path).await, + Err(error) => DiagnosticState::Failed { + document_version: None, + error: format!("synchronizing document: {error:#}"), + }, + } + } + _ => first, + } + } Err(error) => DiagnosticState::Failed { document_version: None, error: format!("reading document: {error}"), @@ -642,6 +695,16 @@ fn publication_matches_document(published: &PublishedDiagnostics, document_versi } } +/// Transport deaths that should trigger an immediate respawn+retry rather than +/// bubbling a permanent failure for the whole batch. +fn is_recoverable_transport_error(err: &anyhow::Error) -> bool { + let text = format!("{err:#}").to_ascii_lowercase(); + text.contains("closed the stream") + || text.contains("lost sync") + || text.contains("broken pipe") + || text.contains("connection reset") +} + /// FNV-1a 64-bit hash. Used only to detect unchanged file contents so we /// can skip redundant `didChange` notifications — not cryptographic. fn fxhash(s: &str) -> u64 { @@ -675,6 +738,22 @@ mod tests { let _ = std::fs::remove_dir_all(root); } + #[test] + fn recoverable_transport_errors_are_detected() { + assert!(is_recoverable_transport_error(&anyhow::anyhow!( + "synchronizing document: LSP server closed the stream" + ))); + assert!(is_recoverable_transport_error(&anyhow::anyhow!( + "LSP stream lost sync during `textDocument/diagnostic`; the server will be restarted on the next query" + ))); + assert!(is_recoverable_transport_error(&anyhow::anyhow!( + "Broken pipe (os error 32)" + ))); + assert!(!is_recoverable_transport_error(&anyhow::anyhow!( + "no LSP server for this file type" + ))); + } + #[test] fn queued_versionless_push_cannot_confirm_clean_after_did_change() { // This models a push queued for version 0 but read only after the