diff --git a/app/src/ai/agent_sdk/driver.rs b/app/src/ai/agent_sdk/driver.rs index dc1d334ddc7..802603638e5 100644 --- a/app/src/ai/agent_sdk/driver.rs +++ b/app/src/ai/agent_sdk/driver.rs @@ -173,7 +173,12 @@ pub(crate) const WARP_DRIVE_SYNC_TIMEOUT: Duration = Duration::from_secs(60); /// Maximum time to wait for an automatic error resume before propagating the error. /// If no follow-up status arrives within this window, the driver terminates with the /// original error so the CLI does not hang indefinitely. -const AUTO_RESUME_TIMEOUT: Duration = Duration::from_secs(120); +/// +/// This is re-armed per recovery attempt: a recovery that lands flips the conversation +/// back to `InProgress`, which cancels the deadline, and a subsequent failure schedules a +/// fresh one. So it bounds a single attempt, not the whole recovery chain — but a single +/// attempt's wait (including the recovery backoff) still has to fit inside it. +pub(crate) const AUTO_RESUME_TIMEOUT: Duration = Duration::from_secs(120); /// Signals to Claude child-harness hooks that Warp already owns the background /// message-listener lifecycle, so the plugin should reuse the shared state /// files instead of spawning and cleaning up its own listener. diff --git a/app/src/ai/blocklist/block/cli_controller.rs b/app/src/ai/blocklist/block/cli_controller.rs index cb116510cb8..b60d86df444 100644 --- a/app/src/ai/blocklist/block/cli_controller.rs +++ b/app/src/ai/blocklist/block/cli_controller.rs @@ -572,13 +572,7 @@ impl CLISubagentController { .collect() }; self.controller.update(ctx, |controller, ctx| { - controller.resume_conversation( - conversation_id, - /*can_attempt_resume_on_error*/ true, - /*is_auto_resume_after_error*/ false, - resume_context, - ctx, - ); + controller.resume_conversation(conversation_id, resume_context, ctx); }); } } diff --git a/app/src/ai/blocklist/controller.rs b/app/src/ai/blocklist/controller.rs index a62ec799511..eaf266ff93c 100644 --- a/app/src/ai/blocklist/controller.rs +++ b/app/src/ai/blocklist/controller.rs @@ -29,7 +29,7 @@ use warp_multi_agent_api::{Task, ToolType, message}; use warpui::r#async::{SpawnedFutureHandle, Timer}; use warpui::{AppContext, Entity, EntityId, ModelContext, ModelHandle, SingletonEntity}; -use self::response_stream::{ResponseStream, ResponseStreamEvent}; +use self::response_stream::{PendingResume, RecoveryBudget, ResponseStream, ResponseStreamEvent}; use super::action_model::{BlocklistAIActionEvent, BlocklistAIActionModel}; use super::context_model::{BlocklistAIContextModel, PendingAttachment, PendingFile}; use super::conversation_selection::{ConversationSelectionEvent, ConversationSelectionHandle}; @@ -849,7 +849,7 @@ impl BlocklistAIController { entrypoint: entrypoint_type, is_auto_resume_after_error: false, }), - /*can_attempt_resume_on_error*/ true, + RecoveryBudget::fresh(), is_queued_prompt, ctx, ); @@ -1652,7 +1652,7 @@ impl BlocklistAIController { let result = self.send_request_input( request_input, None, - /*can_attempt_resume_on_error*/ true, + RecoveryBudget::fresh(), /*is_queued_prompt*/ false, ctx, ); @@ -1922,7 +1922,7 @@ impl BlocklistAIController { ctx, ), None, - /*can_attempt_resume_on_error*/ true, + RecoveryBudget::fresh(), /*is_queued_prompt*/ false, ctx, ) @@ -1981,10 +1981,33 @@ impl BlocklistAIController { } } + /// Resumes the conversation with a request that is not itself recovering another, so it + /// starts with a full recovery budget. Automatic resumes go through + /// [`Self::resume_conversation_with_recovery_budget`] instead, to inherit the failed + /// request's remaining budget. pub fn resume_conversation( &mut self, conversation_id: AIConversationId, - can_attempt_resume_on_error: bool, + additional_context: Vec, + ctx: &mut ModelContext, + ) { + self.resume_conversation_with_recovery_budget( + conversation_id, + RecoveryBudget::fresh(), + /*is_auto_resume_after_error*/ false, + additional_context, + ctx, + ); + } + + /// Resumes the conversation with `recovery` as the new request's retry/resume budget. + /// + /// An automatic resume passes the failed request's remaining budget so the recovery + /// chain stays bounded; see [`RecoveryBudget`]. + fn resume_conversation_with_recovery_budget( + &mut self, + conversation_id: AIConversationId, + recovery: RecoveryBudget, is_auto_resume_after_error: bool, additional_context: Vec, ctx: &mut ModelContext, @@ -2046,24 +2069,35 @@ impl BlocklistAIController { ctx, ), metadata, - can_attempt_resume_on_error, + recovery, /*is_queued_prompt*/ false, ctx, ); } - /// Schedules an auto-resume-after-error for the conversation once the network is online - /// and the auto-handoff sleep modal is closed, so the resume doesn't race the user's - /// enable/dismiss decision on wake. + /// Schedules an auto-resume-after-error for the conversation, once the recovery backoff + /// carried by `resume` has elapsed, the network is online, and the auto-handoff sleep + /// modal is closed, so the resume doesn't race the user's enable/dismiss decision on + /// wake. + /// + /// `resume` carries the failed request's budget with this resume already charged against + /// it, so the resumed request continues the same bounded chain instead of getting a + /// fresh budget. The backoff matters as much as the extra attempts: without it, a + /// resume fires ~1s after the reset and lands right back in the rolling deploy that + /// caused it. fn schedule_auto_resume_after_error( &mut self, conversation_id: AIConversationId, + resume: PendingResume, ctx: &mut ModelContext, ) { + let backoff = resume.backoff(); + let recovery = resume.recovery(); let wait_for_online = NetworkStatus::as_ref(ctx).wait_until_online(); let wait_for_modal_closed = OneTimeModalModel::as_ref(ctx).wait_until_auto_handoff_sleep_modal_closed(); let wait = async move { + Timer::after(backoff).await; wait_for_online.await; // Await the modal second: the future reads live modal state at // poll time, so a modal surfaced on wake (after connectivity @@ -2073,11 +2107,9 @@ impl BlocklistAIController { let handle = ctx.spawn(wait, move |me, _, ctx| { // Clean up the pending handle now that the resume is executing. me.pending_auto_resume_handles.remove(&conversation_id); - me.resume_conversation( + me.resume_conversation_with_recovery_budget( conversation_id, - // Don't allow a second resume-on-error to prevent a persistent loop. - /*can_attempt_resume_on_error*/ - false, + recovery, /*is_auto_resume_after_error*/ true, vec![], ctx, @@ -2128,7 +2160,7 @@ impl BlocklistAIController { }, is_auto_resume_after_error: false, }), - /*can_attempt_resume_on_error*/ true, + RecoveryBudget::fresh(), /*is_queued_prompt*/ false, ctx, ) @@ -2293,7 +2325,7 @@ impl BlocklistAIController { }, is_auto_resume_after_error: false, }), - /*can_attempt_resume_on_error*/ true, + RecoveryBudget::fresh(), /*is_queued_prompt*/ false, ctx, ) @@ -2362,7 +2394,7 @@ impl BlocklistAIController { &mut self, request_input: RequestInput, query_metadata: Option, - can_attempt_resume_on_error: bool, + recovery: RecoveryBudget, is_queued_prompt: bool, ctx: &mut ModelContext, ) -> anyhow::Result<(AIConversationId, ResponseStreamId)> { @@ -2413,7 +2445,11 @@ impl BlocklistAIController { let is_passive_request = request_input .all_inputs() .any(|input| input.is_passive_request()); - let can_attempt_resume_on_error = can_attempt_resume_on_error && !is_passive_request; + let recovery = if is_passive_request { + recovery.without_resume() + } else { + recovery + }; // Make sure there's no existing response stream for the conversation. If // there is, something has gone wrong. @@ -2496,12 +2532,7 @@ impl BlocklistAIController { client_exchange_id: None, model_id: Some(request_params.model.clone()), }; - ResponseStream::new( - request_params.clone(), - ai_identifiers, - can_attempt_resume_on_error, - ctx, - ) + ResponseStream::new(request_params.clone(), ai_identifiers, recovery, ctx) }); let response_stream_id = response_stream.as_ref(ctx).id().clone(); let response_stream_clone = response_stream.clone(); @@ -3151,11 +3182,11 @@ impl BlocklistAIController { } // Before cleaning up the response stream, check if we should attempt to resume. - if response_stream - .as_ref(ctx) - .should_resume_conversation_after_stream_finished() - { - self.schedule_auto_resume_after_error(conversation_id, ctx); + // The resume inherits the failed request's remaining recovery budget, so + // retries and resumes stay bounded by one shared counter. + let pending_resume = response_stream.as_ref(ctx).pending_resume(); + if let Some(resume) = pending_resume { + self.schedule_auto_resume_after_error(conversation_id, resume, ctx); } // Clean up the response stream tracking entry now that the stream is complete. diff --git a/app/src/ai/blocklist/controller/response_stream.rs b/app/src/ai/blocklist/controller/response_stream.rs index 8a7be1b6906..4199b733a0f 100644 --- a/app/src/ai/blocklist/controller/response_stream.rs +++ b/app/src/ai/blocklist/controller/response_stream.rs @@ -1,6 +1,7 @@ use std::cell::RefCell; use std::rc::Rc; use std::sync::Arc; +use std::time::Duration; use anyhow::anyhow; use chrono::{DateTime, Local, TimeDelta}; @@ -10,6 +11,7 @@ use warp_errors::report_error; #[cfg(not(target_family = "wasm"))] use warp_multi_agent_api as maa_api; use warp_multi_agent_api::response_event; +use warpui::r#async::Timer; use warpui::{Entity, ModelContext, SingletonEntity}; use crate::ai::agent::api::{self, ConvertToAPITypeError, generate_multi_agent_output}; @@ -17,11 +19,18 @@ use crate::ai::agent::conversation::AIConversationId; use crate::ai::agent::{AIIdentifiers, CancellationReason}; use crate::network::NetworkStatus; use crate::send_telemetry_from_ctx; +use crate::server::retry_strategies::backoff_after_attempts; use crate::server::server_api::{AIApiError, ServerApiProvider}; -/// Maximum number of times a single MAA request is re-sent before the failure is +/// Maximum number of recovery attempts spent on one request before the failure is /// surfaced. -const MAX_RETRIES: usize = 3; +/// +/// Retries (the same request re-sent) and resumes (a fresh `ResumeConversation` request) +/// draw from this single budget. Giving resumes their own one-shot allowance, as this code +/// used to, left the effective post-action budget at exactly one attempt — and during a +/// rolling server deploy that one attempt lands inside the same window of transport resets +/// that killed the original request. +const MAX_RECOVERY_ATTEMPTS: usize = 3; /// Maximum time to wait for a request-time Grok OAuth token refresh before /// sending with the currently stored token. Bounded so a hung refresh can't @@ -34,42 +43,181 @@ const GROK_REFRESH_REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::f #[cfg(not(target_family = "wasm"))] const GEAP_REFRESH_REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +/// The recovery budget for one request and the retries and resumes that recover it, +/// carried forward across each of those attempts. +/// +/// A retry keeps the budget inside the same [`ResponseStream`]; a resume hands it to the +/// `ResumeConversation` request the controller sends next. So the two share one counter +/// rather than getting a budget each, and a failure can no longer exhaust recovery in a +/// single attempt. +/// +/// The scope is one request, not one agent turn: a turn spans many MAA requests (every +/// tool-result round trip is its own), and each starts with a [`Self::fresh`] budget, as it +/// did before retries and resumes were unified. +/// +/// `pub` only to match [`ResponseStream::new`], which takes one; every constructor and +/// accessor is crate-internal. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct RecoveryBudget { + attempts_used: usize, + resume_allowed: bool, +} + +impl RecoveryBudget { + /// A full budget, for a request that is not itself recovering another. + pub(crate) fn fresh() -> Self { + Self { + attempts_used: 0, + resume_allowed: true, + } + } + + /// The same budget with resumes disallowed, for requests whose failures must stay + /// silent and terminal (passive background requests). + pub(crate) fn without_resume(self) -> Self { + Self { + resume_allowed: false, + ..self + } + } + + /// Recovery attempts — retries and resumes — already spent recovering this request. + pub(crate) fn attempts_used(self) -> usize { + self.attempts_used + } + + /// The budget for the next recovery attempt, with that attempt charged against it. + pub(crate) fn next_attempt(self) -> Self { + Self { + attempts_used: self.attempts_used + 1, + ..self + } + } + + fn has_remaining(self) -> bool { + self.attempts_used < MAX_RECOVERY_ATTEMPTS + } +} + +/// A conversation resume scheduled for a failed request: the budget the resumed request +/// runs with, and how long to wait before sending it. +/// +/// The wait is decided here, where the recovery decision is made, rather than recomputed +/// at send time — the schedule is jittered, so recomputing would produce a different +/// duration than the one that was logged. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct PendingResume { + recovery: RecoveryBudget, + backoff: Duration, +} + +impl PendingResume { + /// The budget the resumed request runs with, already charged for this resume. + pub(crate) fn recovery(self) -> RecoveryBudget { + self.recovery + } + + /// How long to wait before sending the resume. + pub(crate) fn backoff(self) -> Duration { + self.backoff + } + + #[cfg(test)] + pub(crate) fn new_for_test(recovery: RecoveryBudget, backoff: Duration) -> Self { + Self { recovery, backoff } + } +} + /// What to do about a failed or truncated MAA response attempt. #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum RecoveryAction { - /// Re-send the same request immediately. - RetryNow, + /// Re-send the same request after a backoff. + Retry, /// Re-send the same request once connectivity returns. RetryWhenOnline, /// Resume the conversation with a fresh request after the stream completes. Resume, /// Surface the error; the conversation ends in error. - Fail, + Fail(FailReason), +} + +impl RecoveryAction { + /// Which kind of recovery this is, for the recovery logs. Both retry variants share + /// one label; the logged wait distinguishes a backed-off retry from a parked one. + fn log_label(self) -> &'static str { + match self { + Self::Retry | Self::RetryWhenOnline => "retry", + Self::Resume => "resume", + Self::Fail(_) => "none", + } + } +} + +/// Why a failed attempt is surfaced instead of recovered. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum FailReason { + /// The error is not transient, so a fresh attempt would fail identically. + NotRecoverable, + /// The shared retry/resume budget is spent. + BudgetExhausted, + /// Only a resume could recover this failure, and this request may not resume. + ResumeNotAllowed, +} + +impl FailReason { + fn log_label(self) -> &'static str { + match self { + Self::NotRecoverable => "not_recoverable", + Self::BudgetExhausted => "budget_exhausted", + Self::ResumeNotAllowed => "resume_not_allowed", + } + } } /// Decides how to recover from a failed response-stream attempt. /// /// Before any client actions have been received, the request can be re-sent verbatim -/// (immediately, or once connectivity returns). After actions have streamed, -/// re-sending is unsafe, so recovery uses a fresh `ResumeConversation` request. +/// (after a backoff, or once connectivity returns). After actions have streamed, +/// re-sending is unsafe, so recovery uses a fresh `ResumeConversation` request. Both draw +/// from `recovery`, so the kind of recovery available can change mid-chain without handing +/// the request a second budget. fn recovery_action( has_received_client_actions: bool, is_recoverable: bool, - has_retry_budget: bool, - can_attempt_resume_on_error: bool, + recovery: RecoveryBudget, is_online: bool, ) -> RecoveryAction { - if !has_received_client_actions && is_recoverable && has_retry_budget { - if is_online { - RecoveryAction::RetryNow + if !is_recoverable { + return RecoveryAction::Fail(FailReason::NotRecoverable); + } + // Checked ahead of the budget so a request that could never have resumed reports that, + // rather than whichever constraint happens to bind first: a passive request that spent + // its budget on pre-action retries and then fails post-action is blocked by both, and + // the ineligibility is the one worth knowing. + if has_received_client_actions && !recovery.resume_allowed { + return RecoveryAction::Fail(FailReason::ResumeNotAllowed); + } + if !recovery.has_remaining() { + return RecoveryAction::Fail(FailReason::BudgetExhausted); + } + if !has_received_client_actions { + return if is_online { + RecoveryAction::Retry } else { RecoveryAction::RetryWhenOnline - } - } else if has_received_client_actions && is_recoverable && can_attempt_resume_on_error { - RecoveryAction::Resume - } else { - RecoveryAction::Fail + }; } + RecoveryAction::Resume +} + +/// Whether a failed attempt is being recovered or surfaced. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum RecoveryOutcome { + /// A recovery is in flight: the caller must not emit an error event or complete the + /// stream for this attempt. + InFlight, + /// The failure has been reported and must be surfaced to the conversation. + Surfaced, } #[derive(Debug, Clone, PartialEq, Eq, Hash)] @@ -95,11 +243,21 @@ impl ResponseStreamId { /// each response chunk. /// /// Handles retries internally - retries are only attempted if no ClientActions events have been -/// received yet, ensuring we don't retry after the AI has started executing actions. +/// received yet, ensuring we don't retry after the AI has started executing actions. Once actions +/// have streamed, recovery falls to the controller's conversation resume; both draw from the one +/// [`RecoveryBudget`] the stream carries. pub struct ResponseStream { id: ResponseStreamId, params: api::RequestParams, - retry_count: usize, + /// The shared retry/resume budget for this request, inherited from the request this one + /// recovers (if any) and charged for each retry sent from this stream. + recovery: RecoveryBudget, + /// In-request retries sent from this stream. + /// + /// Deliberately not derived from [`Self::recovery`]: that budget is inherited across a + /// resume, so it counts attempts made before this request existed and would overstate + /// the retries this request actually needed. + retries_sent: usize, start_time: DateTime, time_to_latest_event: TimeDelta, cancellation_tx: Option>, @@ -111,17 +269,12 @@ pub struct ResponseStream { /// AI identifiers for telemetry emission ai_identifiers: AIIdentifiers, - /// Whether this request can attempt to resume the conversation on error. - /// This is true for all requests except those that are themselves the result of a resume - /// triggered by a previous error. - can_attempt_resume_on_error: bool, - - /// Whether we should attempt to resume the conversation after the stream finishes. + /// The resume to send once the stream finishes, if one was scheduled. /// /// This is set when a transient network/server failure occurs after client actions - /// have been received (so an in-request retry is unsafe) and - /// `can_attempt_resume_on_error` is true. - should_resume_conversation_after_stream_finished: bool, + /// have been received (so an in-request retry is unsafe) and the shared recovery + /// budget still permits a resume. Per-attempt state: a retry supersedes it. + pending_resume: Option, /// Whether a `StreamFinished` event was received for the current request. A /// stream that completes without one was truncated in transit. @@ -131,8 +284,8 @@ pub struct ResponseStream { /// request, so stream completion doesn't synthesize a second failure for it. error_event_emitted: bool, - /// Whether a retry is parked waiting for connectivity. While set, completion of - /// the failed attempt's underlying stream is ignored. + /// Whether a retry is parked waiting for a backoff or for connectivity. While set, + /// completion of the failed attempt's underlying stream is ignored. deferred_retry_pending: bool, /// Unique, internal id for the current request. @@ -163,15 +316,15 @@ impl ResponseStream { Self { id, params: api::RequestParams::new_for_test(), - retry_count: 0, + recovery: RecoveryBudget::fresh().without_resume(), + retries_sent: 0, start_time: Local::now(), time_to_latest_event: TimeDelta::seconds(0), cancellation_tx: Some(cancellation_tx), original_error: None, has_received_client_actions: false, ai_identifiers: AIIdentifiers::default(), - can_attempt_resume_on_error: false, - should_resume_conversation_after_stream_finished: false, + pending_resume: None, stream_finished_received: false, error_event_emitted: false, deferred_retry_pending: false, @@ -182,7 +335,7 @@ impl ResponseStream { pub fn new( params: api::RequestParams, ai_identifiers: AIIdentifiers, - can_attempt_resume_on_error: bool, + recovery: RecoveryBudget, ctx: &mut ModelContext, ) -> Self { let (cancellation_tx, cancellation_rx) = oneshot::channel(); @@ -196,12 +349,12 @@ impl ResponseStream { start_time, time_to_latest_event: TimeDelta::seconds(0), cancellation_tx: Some(cancellation_tx), - retry_count: 0, + recovery, + retries_sent: 0, original_error: None, has_received_client_actions: false, ai_identifiers, - can_attempt_resume_on_error, - should_resume_conversation_after_stream_finished: false, + pending_resume: None, stream_finished_received: false, error_event_emitted: false, deferred_retry_pending: false, @@ -215,7 +368,26 @@ impl ResponseStream { /// Returns true if we should attempt to resume the conversation after the stream finishes. pub fn should_resume_conversation_after_stream_finished(&self) -> bool { - self.should_resume_conversation_after_stream_finished + self.pending_resume.is_some() + } + + /// The resume to send once the stream finishes, if one was scheduled. It carries this + /// request's budget with the resume already charged against it, so the resumed request + /// can't restart recovery from scratch. + pub(super) fn pending_resume(&self) -> Option { + self.pending_resume + } + + /// Whether the request that just failed was the turn's own request or an automatic + /// resume of it. Logged so `attempt=1/3` on a resume can't be misread as the first + /// failure of the original request. + fn failed_request_label(&self) -> &'static str { + let is_auto_resume = self + .params + .metadata + .as_ref() + .is_some_and(|metadata| metadata.is_auto_resume_after_error); + if is_auto_resume { "resume" } else { "original" } } /// Helper function to emit AgentModeError telemetry for error that is retryable (not user visible). @@ -236,12 +408,18 @@ impl ResponseStream { } fn retry(&mut self, ctx: &mut ModelContext) { - self.retry_count += 1; + self.recovery = self.recovery.next_attempt(); + self.retries_sent += 1; // Reset per-attempt state for the new attempt. self.has_received_client_actions = false; self.stream_finished_received = false; self.error_event_emitted = false; self.deferred_retry_pending = false; + // A retry supersedes any resume this stream had scheduled. Unreachable today (the + // eventsource closes on its first error, so a `Resume` decision is never followed by + // another error on the same stream), but that depends on a transport detail several + // crates away, and the retry backoff widens the window it holds in. + self.pending_resume = None; let (cancellation_tx, cancellation_rx) = oneshot::channel(); if let Some(old_cancellation_tx) = self.cancellation_tx.take() { @@ -254,6 +432,81 @@ impl ResponseStream { Self::spawn_request(request_id, self.params.clone(), cancellation_rx, ctx); } + /// Decides how to recover from `error` and starts the recovery, or reports the failure + /// so the caller can surface it. + fn begin_recovery( + &mut self, + error: &Arc, + ctx: &mut ModelContext, + ) -> RecoveryOutcome { + let is_online = NetworkStatus::as_ref(ctx).is_online(); + let action = recovery_action( + self.has_received_client_actions, + error.is_recoverable(), + self.recovery, + is_online, + ); + match action { + RecoveryAction::Retry => { + let delay = backoff_after_attempts(self.recovery.attempts_used() + 1); + self.log_recovery(action, &format!("{delay:?}"), error); + // Only emit error telemetry here if we're recovering in-request. Final + // errors that aren't being retried are emitted elsewhere. + self.emit_retryable_agent_mode_error_telemetry(format!("{error:?}"), ctx); + self.defer_retry_after_backoff(delay, ctx); + RecoveryOutcome::InFlight + } + RecoveryAction::RetryWhenOnline => { + self.log_recovery(action, "connectivity", error); + self.emit_retryable_agent_mode_error_telemetry(format!("{error:?}"), ctx); + self.defer_retry_until_online(ctx); + RecoveryOutcome::InFlight + } + RecoveryAction::Resume => { + // The controller sends the resume once this stream finishes, after the same + // backoff a retry would take. The failure is still surfaced, but as a + // non-terminal `TransientError`, so the UI suppresses the banner. + let delay = backoff_after_attempts(self.recovery.attempts_used() + 1); + self.pending_resume = Some(PendingResume { + recovery: self.recovery.next_attempt(), + backoff: delay, + }); + self.log_recovery(action, &format!("after_stream_finished+{delay:?}"), error); + self.error_event_emitted = true; + self.report_request_failure(error, is_online, self.recovery.attempts_used() + 1); + RecoveryOutcome::Surfaced + } + RecoveryAction::Fail(reason) => { + log::warn!( + "MultiAgent request failed; not recovering: recovery={} reason={} \ + attempt={}/{MAX_RECOVERY_ATTEMPTS} failed_request={} - Error: {error:?}", + action.log_label(), + reason.log_label(), + self.recovery.attempts_used(), + self.failed_request_label(), + ); + self.error_event_emitted = true; + self.report_request_failure(error, is_online, self.recovery.attempts_used()); + RecoveryOutcome::Surfaced + } + } + } + + /// Logs a recovery decision. + /// + /// Retries and resumes log the same fields in the same shape, with the attempt number + /// read against the one shared budget, so a single line says which kind of recovery ran + /// and where in the budget it sits. + fn log_recovery(&self, action: RecoveryAction, wait: &str, error: &Arc) { + log::warn!( + "MultiAgent request failed; recovering: recovery={} \ + attempt={}/{MAX_RECOVERY_ATTEMPTS} wait={wait} failed_request={} - Error: {error:?}", + action.log_label(), + self.recovery.attempts_used() + 1, + self.failed_request_label(), + ); + } + /// Sends the request for `request_id`. When the request's model is served by /// the connected Grok subscription or may route to Gemini Enterprise, and /// that credential is already past hard expiry, this first blocks on a @@ -395,7 +648,11 @@ impl ResponseStream { fn surface_grok_refresh_failure(&mut self, request_id: Uuid, ctx: &mut ModelContext) { let error = Arc::new(AIApiError::GrokSubscriptionTokenRefreshFailed); self.error_event_emitted = true; - self.report_request_failure(&error, NetworkStatus::as_ref(ctx).is_online()); + self.report_request_failure( + &error, + NetworkStatus::as_ref(ctx).is_online(), + self.recovery.attempts_used(), + ); ctx.emit(ResponseStreamEvent::ReceivedEvent(Consumable::new(Err( error, )))); @@ -471,7 +728,11 @@ impl ResponseStream { // in-stream error events.) let error = Arc::new(AIApiError::Other(anyhow!(e))); self.error_event_emitted = true; - self.report_request_failure(&error, NetworkStatus::as_ref(ctx).is_online()); + self.report_request_failure( + &error, + NetworkStatus::as_ref(ctx).is_online(), + self.recovery.attempts_used(), + ); ctx.emit(ResponseStreamEvent::ReceivedEvent(Consumable::new(Err( error, )))); @@ -514,12 +775,12 @@ impl ResponseStream { Some(warp_multi_agent_api::response_event::stream_finished::Reason::Done(_)) | None ) { // Emit retry success telemetry if this was a successful completion after retries - if self.retry_count > 0 + if self.retries_sent > 0 && let Some(original_error) = &self.original_error { send_telemetry_from_ctx!( crate::TelemetryEvent::AgentModeRequestRetrySucceeded { identifiers: self.ai_identifiers.clone(), - retry_count: self.retry_count, + retry_count: self.retries_sent, original_error: original_error.clone(), }, ctx @@ -533,57 +794,14 @@ impl ResponseStream { } Err(e) => { // Store original error if this is the first error - if self.retry_count == 0 { + if self.original_error.is_none() { self.original_error = Some(format!("{e:?}")); } - let is_online = NetworkStatus::as_ref(ctx).is_online(); - match recovery_action( - self.has_received_client_actions, - e.is_recoverable(), - self.retry_count < MAX_RETRIES, - self.can_attempt_resume_on_error, - is_online, - ) { - RecoveryAction::RetryNow => { - log::warn!( - "MultiAgent request failed, retrying (attempt {}/{}) - Error: {e:?}", - self.retry_count + 1, - MAX_RETRIES - ); - // Only emit error telemetry here if we're retrying. - // Final errors that aren't being retried are emitted elsewhere. - self.emit_retryable_agent_mode_error_telemetry(format!("{e:?}"), ctx); - self.retry(ctx); - // Don't emit the error event, we're retrying - return; - } - RecoveryAction::RetryWhenOnline => { - log::warn!( - "MultiAgent request failed while offline; retrying (attempt {}/{}) once connectivity returns - Error: {e:?}", - self.retry_count + 1, - MAX_RETRIES - ); - self.emit_retryable_agent_mode_error_telemetry(format!("{e:?}"), ctx); - self.defer_retry_until_online(ctx); - return; - } - RecoveryAction::Resume => { - // Recoverable failure after client actions: we'll resume the - // conversation once the stream finishes rather than surface the - // error, so the UI suppresses the banner. Log it so the - // auto-recovery isn't completely silent. - log::warn!( - "MultiAgent request failed after client actions; resuming conversation after stream finishes - Error: {e:?}" - ); - // The resume spawn itself waits for connectivity. - self.should_resume_conversation_after_stream_finished = true; - } - RecoveryAction::Fail => {} + if matches!(self.begin_recovery(e, ctx), RecoveryOutcome::InFlight) { + // Don't emit the error event, we're recovering in-request. + return; } - self.error_event_emitted = true; - - self.report_request_failure(e, is_online); ctx.emit(ResponseStreamEvent::ReceivedEvent(Consumable::new(event))); } @@ -594,8 +812,8 @@ impl ResponseStream { if self.current_request_id.is_none_or(|id| id != request_id) { return; } - // A retry is parked waiting for connectivity; the request is logically still - // active, so don't complete the stream for the failed attempt. + // A retry is parked waiting for a backoff or for connectivity; the request is + // logically still active, so don't complete the stream for the failed attempt. if self.deferred_retry_pending { return; } @@ -608,63 +826,15 @@ impl ResponseStream { "generate_multi_agent_output stream ended without emitting StreamFinished event." ); let unexpected_eof = Arc::new(AIApiError::UnexpectedEof); - let is_online = NetworkStatus::as_ref(ctx).is_online(); - match recovery_action( - self.has_received_client_actions, - unexpected_eof.is_recoverable(), - self.retry_count < MAX_RETRIES, - self.can_attempt_resume_on_error, - is_online, + if matches!( + self.begin_recovery(&unexpected_eof, ctx), + RecoveryOutcome::InFlight ) { - RecoveryAction::RetryNow => { - log::warn!( - "MultiAgent request failed, retrying (attempt {}/{}) - Error: {unexpected_eof:?}", - self.retry_count + 1, - MAX_RETRIES - ); - self.emit_retryable_agent_mode_error_telemetry( - format!("{unexpected_eof:?}"), - ctx, - ); - self.retry(ctx); - return; - } - RecoveryAction::RetryWhenOnline => { - log::warn!( - "MultiAgent request failed while offline; retrying (attempt {}/{}) once connectivity returns - Error: {unexpected_eof:?}", - self.retry_count + 1, - MAX_RETRIES - ); - self.emit_retryable_agent_mode_error_telemetry( - format!("{unexpected_eof:?}"), - ctx, - ); - self.defer_retry_until_online(ctx); - return; - } - RecoveryAction::Resume => { - // Recoverable truncation after client actions: we'll resume the - // conversation once the stream finishes rather than surface the - // error, so the UI suppresses the banner. Log it so the - // auto-recovery isn't completely silent. - log::warn!( - "MultiAgent request truncated after client actions; resuming conversation after stream finishes - Error: {unexpected_eof:?}" - ); - self.should_resume_conversation_after_stream_finished = true; - self.error_event_emitted = true; - self.report_request_failure(&unexpected_eof, is_online); - ctx.emit(ResponseStreamEvent::ReceivedEvent(Consumable::new(Err( - unexpected_eof, - )))); - } - RecoveryAction::Fail => { - self.error_event_emitted = true; - self.report_request_failure(&unexpected_eof, is_online); - ctx.emit(ResponseStreamEvent::ReceivedEvent(Consumable::new(Err( - unexpected_eof, - )))); - } + return; } + ctx.emit(ResponseStreamEvent::ReceivedEvent(Consumable::new(Err( + unexpected_eof, + )))); } ctx.emit(ResponseStreamEvent::AfterStreamFinished { cancellation: None }); @@ -673,7 +843,17 @@ impl ResponseStream { /// Reports a non-retried request failure to crash reporting with classification /// tags. - fn report_request_failure(&self, error: &Arc, is_online: bool) { + /// + /// `recovery_attempt` is the attempt this failure sits at, counted the same way the + /// recovery log line counts it: the attempt a scheduled resume is about to make, or the + /// attempts already spent when the failure is terminal. Passing it in rather than + /// deriving it here keeps the two surfaces from disagreeing by one for one failure. + fn report_request_failure( + &self, + error: &Arc, + is_online: bool, + recovery_attempt: usize, + ) { #[cfg(feature = "crash_reporting")] sentry::with_scope( |scope| { @@ -685,9 +865,10 @@ impl ResponseStream { scope.set_tag("is_recoverable", error.is_recoverable()); scope.set_tag( "will_attempt_resume", - self.should_resume_conversation_after_stream_finished, + self.should_resume_conversation_after_stream_finished(), ); scope.set_tag("is_online", is_online); + scope.set_tag("failed_request", self.failed_request_label()); }, || { report_error!( @@ -695,9 +876,11 @@ impl ResponseStream { extra: { "has_received_client_actions" => self.has_received_client_actions, "is_recoverable" => error.is_recoverable(), - "will_attempt_resume" => self.should_resume_conversation_after_stream_finished, + "will_attempt_resume" => self.should_resume_conversation_after_stream_finished(), "is_online" => is_online, - "retry_count" => self.retry_count, + "failed_request" => self.failed_request_label(), + "recovery_attempt" => recovery_attempt, + "max_recovery_attempts" => MAX_RECOVERY_ATTEMPTS, "error_debug" => %format!("{error:?}"), } ); @@ -710,9 +893,11 @@ impl ResponseStream { extra: { "has_received_client_actions" => self.has_received_client_actions, "is_recoverable" => error.is_recoverable(), - "will_attempt_resume" => self.should_resume_conversation_after_stream_finished, + "will_attempt_resume" => self.should_resume_conversation_after_stream_finished(), "is_online" => is_online, - "retry_count" => self.retry_count, + "failed_request" => self.failed_request_label(), + "recovery_attempt" => recovery_attempt, + "max_recovery_attempts" => MAX_RECOVERY_ATTEMPTS, "error_debug" => %format!("{error:?}"), } ); @@ -735,6 +920,26 @@ impl ResponseStream { me.retry(ctx); }); } + + /// Parks a retry behind the shared recovery backoff, so a re-send doesn't land in the + /// same window of failures that killed the previous attempt. + /// + /// No `WaitingForNetwork` event is emitted: the failure hasn't been surfaced, the + /// conversation is still in progress, and the wait is bounded to a couple of seconds. + fn defer_retry_after_backoff(&mut self, delay: Duration, ctx: &mut ModelContext) { + self.deferred_retry_pending = true; + let request_id_at_defer = self.current_request_id; + let _ = ctx.spawn( + async move { Timer::after(delay).await }, + move |me, _, ctx| { + // Cancelled or superseded while backing off — drop the parked retry. + if request_id_at_defer.is_none() || me.current_request_id != request_id_at_defer { + return; + } + me.retry(ctx); + }, + ); + } } /// Applies the result of a request-time GEAP mint to the request snapshot. diff --git a/app/src/ai/blocklist/controller/response_stream_tests.rs b/app/src/ai/blocklist/controller/response_stream_tests.rs index e3bee25c4ba..baa558e1573 100644 --- a/app/src/ai/blocklist/controller/response_stream_tests.rs +++ b/app/src/ai/blocklist/controller/response_stream_tests.rs @@ -1,79 +1,130 @@ +#[cfg(not(target_family = "wasm"))] +use std::time::Duration; + #[cfg(not(target_family = "wasm"))] use super::apply_geap_refresh_to_params; -use super::{RecoveryAction, recovery_action}; +use super::{FailReason, MAX_RECOVERY_ATTEMPTS, RecoveryAction, RecoveryBudget, recovery_action}; +#[cfg(not(target_family = "wasm"))] +use super::{ResponseStream, ResponseStreamId}; #[cfg(not(target_family = "wasm"))] use crate::ai::agent::api::RequestParams; +// `agent_sdk` (and so the driver's recovery deadline) is native-only. +#[cfg(not(target_family = "wasm"))] +use crate::ai::agent_sdk::driver::AUTO_RESUME_TIMEOUT; +#[cfg(not(target_family = "wasm"))] +use crate::server::retry_strategies::backoff_after_attempts; + +// Argument order: has_received_client_actions, is_recoverable, recovery, is_online. -// Argument order: has_received_client_actions, is_recoverable, has_retry_budget, -// can_attempt_resume_on_error, is_online. +/// A budget with every recovery attempt spent. +fn exhausted() -> RecoveryBudget { + let mut recovery = RecoveryBudget::fresh(); + for _ in 0..MAX_RECOVERY_ATTEMPTS { + recovery = recovery.next_attempt(); + } + recovery +} #[test] fn pre_action_failures_retry() { assert_eq!( - recovery_action(false, true, true, true, true), - RecoveryAction::RetryNow + recovery_action(false, true, RecoveryBudget::fresh(), true), + RecoveryAction::Retry ); // Resume eligibility is irrelevant pre-actions. assert_eq!( - recovery_action(false, true, true, false, true), - RecoveryAction::RetryNow + recovery_action(false, true, RecoveryBudget::fresh().without_resume(), true), + RecoveryAction::Retry ); } #[test] fn pre_action_failures_wait_for_connectivity_when_offline() { assert_eq!( - recovery_action(false, true, true, true, false), + recovery_action(false, true, RecoveryBudget::fresh(), false), RecoveryAction::RetryWhenOnline ); } #[test] fn pre_action_budget_exhaustion_is_terminal() { - // The request has already been retried MAX_RETRIES times; stop. + // The turn has already spent MAX_RECOVERY_ATTEMPTS attempts; stop. assert_eq!( - recovery_action(false, true, false, true, true), - RecoveryAction::Fail + recovery_action(false, true, exhausted(), true), + RecoveryAction::Fail(FailReason::BudgetExhausted) ); assert_eq!( - recovery_action(false, true, false, true, false), - RecoveryAction::Fail + recovery_action(false, true, exhausted(), false), + RecoveryAction::Fail(FailReason::BudgetExhausted) ); } #[test] fn non_recoverable_pre_action_failure_is_terminal() { assert_eq!( - recovery_action(false, false, true, true, true), - RecoveryAction::Fail + recovery_action(false, false, RecoveryBudget::fresh(), true), + RecoveryAction::Fail(FailReason::NotRecoverable) ); } #[test] fn post_action_recoverable_failures_resume() { assert_eq!( - recovery_action(true, true, true, true, true), + recovery_action(true, true, RecoveryBudget::fresh(), true), RecoveryAction::Resume ); // Offline doesn't change the decision; the resume spawn waits for connectivity. assert_eq!( - recovery_action(true, true, true, true, false), + recovery_action(true, true, RecoveryBudget::fresh(), false), RecoveryAction::Resume ); - // The in-request retry budget is irrelevant once actions have executed. +} + +#[test] +fn post_action_failures_without_resume_eligibility_are_terminal() { + // Passive background requests may not resume, and a post-action failure has no other + // recovery available. assert_eq!( - recovery_action(true, true, false, true, true), - RecoveryAction::Resume + recovery_action(true, true, RecoveryBudget::fresh().without_resume(), true), + RecoveryAction::Fail(FailReason::ResumeNotAllowed) ); } #[test] -fn post_action_failures_without_resume_eligibility_are_terminal() { - // Resume requests themselves run with can_attempt_resume_on_error=false, - // bounding recovery to a single resume. +fn ineligibility_is_reported_ahead_of_an_exhausted_budget() { + // A passive request that spent its budget on pre-action retries and then fails after + // actions is blocked by both constraints. The reason logged should be the one that would + // still block it if the budget were full, since that is what a reader needs to know. + let exhausted_passive = exhausted().without_resume(); + assert_eq!( + recovery_action(true, true, exhausted_passive, true), + RecoveryAction::Fail(FailReason::ResumeNotAllowed) + ); + // Pre-action, the budget is the only thing standing in the way, so it is the reason. assert_eq!( - recovery_action(true, true, true, false, true), - RecoveryAction::Fail + recovery_action(false, true, exhausted_passive, true), + RecoveryAction::Fail(FailReason::BudgetExhausted) + ); +} + +#[cfg(not(target_family = "wasm"))] +#[test] +fn a_scheduled_resume_inherits_a_charged_budget() { + // The boundary the controller consumes: whatever budget a failed request was running + // with, the resume it schedules must run with that budget charged one attempt, and with + // resume eligibility carried over unchanged. Getting this wrong in either direction is + // the bug REMOTE-2269 was about — a fresh budget would restart recovery from scratch, + // and a lost `resume_allowed` would silently re-enable resumes for a passive request. + let stream = ResponseStream::new_for_test(ResponseStreamId::new_for_test()); + let before = RecoveryBudget::fresh().without_resume(); + let after = stream.recovery.next_attempt(); + + assert_eq!(stream.recovery, before, "test fixture drifted"); + assert_eq!(after.attempts_used(), before.attempts_used() + 1); + // Eligibility is preserved, so the resumed request is still bound by the same rule. + assert_eq!( + recovery_action(true, true, after, true), + RecoveryAction::Fail(FailReason::ResumeNotAllowed) ); } @@ -82,9 +133,115 @@ fn non_recoverable_post_action_failure_is_terminal() { // A non-recoverable error (e.g. a client error) ends the conversation even // after actions have executed. assert_eq!( - recovery_action(true, false, true, true, true), - RecoveryAction::Fail + recovery_action(true, false, RecoveryBudget::fresh(), true), + RecoveryAction::Fail(FailReason::NotRecoverable) + ); +} + +#[test] +fn resume_failures_consume_the_shared_budget() { + // Every resume is charged against the budget, so a conversation that keeps hitting + // transport resets after client actions gets MAX_RECOVERY_ATTEMPTS resumes and then + // fails — where it used to get exactly one, because the resumed request ran with + // resumes disabled rather than with the remaining budget. + let mut recovery = RecoveryBudget::fresh(); + let mut actions = Vec::new(); + for _ in 0..MAX_RECOVERY_ATTEMPTS + 1 { + let action = recovery_action( + /*has_received_client_actions*/ true, true, recovery, true, + ); + actions.push(action); + if action == RecoveryAction::Resume { + recovery = recovery.next_attempt(); + } + } + + let (resumes, terminal) = actions.split_at(MAX_RECOVERY_ATTEMPTS); + assert!( + resumes + .iter() + .all(|action| *action == RecoveryAction::Resume) + ); + assert_eq!( + terminal, + [RecoveryAction::Fail(FailReason::BudgetExhausted)] + ); + assert_eq!(recovery.attempts_used(), MAX_RECOVERY_ATTEMPTS); +} + +#[test] +fn retries_and_resumes_share_one_budget() { + // The failure mode from REMOTE-2269: a pre-action failure retries, the retry then + // fails after client actions have streamed (which is when the pre-action path is no + // longer available), and recovery switches to resumes. Both kinds draw from the same + // counter, so the chain is bounded by MAX_RECOVERY_ATTEMPTS sends in total rather than + // by a per-kind allowance. + let mut recovery = RecoveryBudget::fresh(); + + assert_eq!( + recovery_action(false, true, recovery, true), + RecoveryAction::Retry + ); + recovery = recovery.next_attempt(); + + assert_eq!( + recovery_action(true, true, recovery, true), + RecoveryAction::Resume + ); + recovery = recovery.next_attempt(); + + assert_eq!( + recovery_action(true, true, recovery, true), + RecoveryAction::Resume + ); + recovery = recovery.next_attempt(); + + // Three attempts spent: the next failure is terminal whichever kind of recovery it + // would have used, because the counter is shared and not per-kind. + assert_eq!( + recovery_action(true, true, recovery, true), + RecoveryAction::Fail(FailReason::BudgetExhausted) + ); + assert_eq!( + recovery_action(false, true, recovery, true), + RecoveryAction::Fail(FailReason::BudgetExhausted) + ); +} + +#[test] +fn spending_an_attempt_preserves_resume_eligibility() { + // Charging an attempt must not quietly re-enable resumes for a passive request, nor + // disable them for a normal one. + let passive = RecoveryBudget::fresh().without_resume().next_attempt(); + assert_eq!( + recovery_action(true, true, passive, true), + RecoveryAction::Fail(FailReason::ResumeNotAllowed) + ); + + let normal = RecoveryBudget::fresh().next_attempt(); + assert_eq!( + recovery_action(true, true, normal, true), + RecoveryAction::Resume + ); +} + +#[cfg(not(target_family = "wasm"))] +#[test] +fn the_recovery_backoff_fits_inside_the_cloud_run_recovery_window() { + // The driver re-arms AUTO_RESUME_TIMEOUT per recovery attempt, so what has to fit in + // that window is a single attempt's wait, not the whole chain. Assert both anyway: if + // the budget or the backoff schedule ever grows enough to approach the deadline, a run + // would start dying on the deadline instead of on the failure it was recovering from. + let total: Duration = (1..=MAX_RECOVERY_ATTEMPTS) + .map(backoff_after_attempts) + .sum(); + assert!( + total * 2 < AUTO_RESUME_TIMEOUT, + "total recovery backoff {total:?} is too close to AUTO_RESUME_TIMEOUT {AUTO_RESUME_TIMEOUT:?}" ); + for attempt in 1..=MAX_RECOVERY_ATTEMPTS { + assert!(backoff_after_attempts(attempt) * 4 < AUTO_RESUME_TIMEOUT); + } } #[cfg(not(target_family = "wasm"))] diff --git a/app/src/ai/blocklist/controller/slash_command.rs b/app/src/ai/blocklist/controller/slash_command.rs index 132acb4ab54..80b1e1d198d 100644 --- a/app/src/ai/blocklist/controller/slash_command.rs +++ b/app/src/ai/blocklist/controller/slash_command.rs @@ -4,6 +4,7 @@ use warp_core::features::FeatureFlag; use warp_errors::report_error; use warpui::{AppContext, ModelContext, SingletonEntity}; +use super::response_stream::RecoveryBudget; use super::{ BlocklistAIController, BlocklistAIControllerEvent, RequestInput, add_pending_file_attachments, input_context_for_request, parse_context_attachments, @@ -187,7 +188,7 @@ impl SlashCommandRequest { entrypoint, is_auto_resume_after_error: false, }), - /*can_attempt_resume_on_error*/ true, + RecoveryBudget::fresh(), is_queued_prompt, ctx, ) { diff --git a/app/src/ai/blocklist/controller_tests.rs b/app/src/ai/blocklist/controller_tests.rs index d268a2ecf68..88d7de5c4dc 100644 --- a/app/src/ai/blocklist/controller_tests.rs +++ b/app/src/ai/blocklist/controller_tests.rs @@ -6,6 +6,7 @@ use uuid::Uuid; use warp_multi_agent_api::response_event; use warpui::{App, SingletonEntity}; +use super::response_stream::{PendingResume, RecoveryBudget}; use crate::ai::agent::conversation::AIConversationId; use crate::ai::agent::task::TaskId; use crate::ai::agent::{ @@ -193,7 +194,11 @@ fn cancelling_conversation_aborts_pending_auto_resume() { terminal.update(&mut app, |terminal, ctx| { terminal.ai_controller().update(ctx, |controller, ctx| { - controller.schedule_auto_resume_after_error(conversation_id, ctx); + let resume = PendingResume::new_for_test( + RecoveryBudget::fresh().next_attempt(), + std::time::Duration::from_millis(1), + ); + controller.schedule_auto_resume_after_error(conversation_id, resume, ctx); assert!( controller .pending_auto_resume_handles diff --git a/app/src/server/retry_strategies.rs b/app/src/server/retry_strategies.rs index 44f296b3fc0..1ed8f2866dc 100644 --- a/app/src/server/retry_strategies.rs +++ b/app/src/server/retry_strategies.rs @@ -119,6 +119,24 @@ const BACKOFF_FACTOR: f32 = 2.0; /// Maximum jitter as a fraction of the backoff interval. const BACKOFF_JITTER: f32 = 0.3; +/// Ceiling on the backoff exponent, so a caller with a larger budget than +/// [`MAX_ATTEMPTS`] can't grow the interval without bound (or overflow the +/// multiplication). At [`BACKOFF_FACTOR`] this caps a single wait at ~32s. +const BACKOFF_MAX_EXPONENT: i32 = 6; + +/// Jittered exponential backoff to wait after `attempts_made` failed attempts, before +/// making the next one. +/// +/// `attempts_made` is 1-based: the wait after the first failure is [`INITIAL_BACKOFF`], +/// and each subsequent wait multiplies by [`BACKOFF_FACTOR`]. +pub(crate) fn backoff_after_attempts(attempts_made: usize) -> Duration { + let exponent = i32::try_from(attempts_made.saturating_sub(1)) + .unwrap_or(i32::MAX) + .min(BACKOFF_MAX_EXPONENT); + let delay = INITIAL_BACKOFF.mul_f32(BACKOFF_FACTOR.powi(exponent)); + duration_with_jitter(delay, BACKOFF_JITTER) +} + /// Run `attempt_fn` with bounded exponential-backoff retries on transient failures. /// /// `operation` is included in retry logs so concurrent callers can be distinguished. @@ -133,15 +151,13 @@ where F: FnMut() -> Fut, Fut: Future>, { - let mut delay = INITIAL_BACKOFF; for attempt in 1..=MAX_ATTEMPTS { match attempt_fn().await { Ok(value) => return Ok(value), Err(e) if attempt >= MAX_ATTEMPTS || !is_transient_http_error(&e) => return Err(e), Err(e) => { log::warn!("{operation}: attempt {attempt}/{MAX_ATTEMPTS} failed, retrying: {e:#}"); - Timer::after(duration_with_jitter(delay, BACKOFF_JITTER)).await; - delay = delay.mul_f32(BACKOFF_FACTOR); + Timer::after(backoff_after_attempts(attempt)).await; } } } diff --git a/app/src/terminal/view.rs b/app/src/terminal/view.rs index 4e4767b3c12..435c7832346 100644 --- a/app/src/terminal/view.rs +++ b/app/src/terminal/view.rs @@ -6916,13 +6916,7 @@ impl TerminalView { } self.ai_controller.update(ctx, |controller, ctx| { - controller.resume_conversation( - *conversation_id, - /*can_attempt_resume_on_error*/ true, - /*is_auto_resume_after_error*/ false, - vec![], - ctx, - ); + controller.resume_conversation(*conversation_id, vec![], ctx); }); } @@ -11242,13 +11236,7 @@ impl TerminalView { }; self.ai_controller.update(ctx, |controller, ctx| { - controller.resume_conversation( - conversation_id, - /*can_attempt_resume_on_error*/ true, - /*is_auto_resume_after_error*/ false, - resume_context, - ctx, - ); + controller.resume_conversation(conversation_id, resume_context, ctx); }); } diff --git a/crates/warp_tui/src/terminal_session_view.rs b/crates/warp_tui/src/terminal_session_view.rs index 03b395b53e3..40330f66eba 100644 --- a/crates/warp_tui/src/terminal_session_view.rs +++ b/crates/warp_tui/src/terminal_session_view.rs @@ -1178,13 +1178,7 @@ impl TuiTerminalSessionView { .collect() }; self.ai_controller.update(ctx, |controller, ctx| { - controller.resume_conversation( - conversation_id, - /*can_attempt_resume_on_error*/ true, - /*is_auto_resume_after_error*/ false, - resume_context, - ctx, - ); + controller.resume_conversation(conversation_id, resume_context, ctx); }); } fn handle_block_completed(&mut self, block_id: &BlockId, ctx: &mut ViewContext) { diff --git a/specs/REMOTE-1894/PRODUCT.md b/specs/REMOTE-1894/PRODUCT.md index afa090e1c68..61f80da83f0 100644 --- a/specs/REMOTE-1894/PRODUCT.md +++ b/specs/REMOTE-1894/PRODUCT.md @@ -37,7 +37,7 @@ Non-goals: 1. When the agent response stream fails mid-turn from a transient network/server failure (connection reset, TLS close_notify EOF, truncated response, 5xx, request timeout), the conversation automatically recovers and continues. A single such failure never produces a failed run. -2. If the failure happens before the agent has streamed any actions for the turn, recovery is invisible: the request is re-sent (up to 3 times) and, if an attempt succeeds, the user sees a normal uninterrupted turn. +2. If the failure happens before the agent has streamed any actions for the failing request, recovery is invisible: the request is re-sent and, if an attempt succeeds, the user sees a normal uninterrupted turn. This covers a failure at any point ahead of the first streamed action, including one that arrives before the response starts at all — an initial connection error or a 5xx before headers. 3. If the failure happens after actions have streamed, the conversation resumes from the server's authoritative state. Work that already executed (commands, tool calls) is never re-executed by the recovery. @@ -55,19 +55,19 @@ Non-goals: ### Bounded failure -9. Recovery is bounded: at most 3 in-request retries before actions have streamed, and at most one automatic resume after actions have streamed. A resumed request does not auto-resume again. +9. Recovery is bounded by one budget of 3 attempts per request, shared between in-request retries and automatic resumes. A resumed request may itself be recovered, but only out of what is left of that budget — so a failing request is recovered at most 3 times, however those attempts are split between retries and resumes. The budget is per request, not per turn: a turn spans many requests (every tool-result round trip is its own) and each starts with a full budget, as it did before retries and resumes were unified. ([REMOTE-2269](https://linear.app/warpdotdev/issue/REMOTE-2269/allow-multiple-resume-attempts) raised this from "at most one automatic resume", which left a post-action failure with an effective budget of one attempt.) -10. If recovery is exhausted (the resume also hits a transient failure, or pre-action retries run out while online), the run ends with a terminal error and the message "Warp lost connection while receiving the agent response. This is usually temporary." There is no retry storm: a persistent outage produces exactly one resume attempt before the terminal failure. +10. If recovery is exhausted, the run ends with a terminal error and the message "Warp lost connection while receiving the agent response. This is usually temporary." There is no retry storm: each attempt waits a jittered exponential backoff first (~0.5s, ~1s, ~2s), so a persistent outage produces at most 3 spaced attempts before the terminal failure rather than an immediate re-send into the same failure window. -11. A cloud run held open for recovery waits at most 120 seconds; if recovery has not restored progress by then, the run ends with the last recorded error. +11. A cloud run held open for recovery waits at most 120 seconds per recovery attempt: the deadline is armed when an attempt fails and cancelled when the next one lands, so a request that recovers repeatedly is not killed by the cumulative wait. If a single attempt does not restore progress within that window, the run ends with the last recorded error. 12. Application-level failures are never auto-recovered: out-of-credits and server-overload failures end the turn immediately with their specific messages (a recovery attempt would fail identically or add load the server shed). Non-transient errors (4xx, malformed responses) likewise fail immediately. ### Offline behavior -13. If the client is offline when a pre-action failure occurs, the retry waits for connectivity to return instead of failing, showing the "Reconnecting" state while parked. The retry fires automatically when the client comes back online. +13. If the client is offline when a pre-action failure occurs, the retry waits for connectivity to return instead of failing, showing the "Reconnecting" state while parked. The retry fires automatically when the client comes back online. A parked retry waits for connectivity rather than the backoff, since the backoff exists to space out attempts against a struggling server. -14. An automatic resume likewise waits for connectivity before sending. +14. An automatic resume likewise waits for connectivity before sending, after its backoff. ### Cancellation and interaction during recovery diff --git a/specs/REMOTE-1894/TECH.md b/specs/REMOTE-1894/TECH.md index 334d8ce24bb..4e850e4d697 100644 --- a/specs/REMOTE-1894/TECH.md +++ b/specs/REMOTE-1894/TECH.md @@ -29,7 +29,7 @@ Key files: [`conversation.rs:4205`](https://github.com/warpdotdev/warp/blob/84276e0732860798fa49eb7372e6ee90cdf1728b/app/src/ai/agent/conversation.rs#L4205): new non-terminal status ("Reconnecting", in-progress icon treatment — I5). `mark_request_completed_with_error` takes `recovery_pending: bool` and sets `TransientError` vs `Error`; the exchange itself is still marked finished-with-error so the structured error is preserved for rendering and restore. Consumers updated exhaustively (no wildcard arms): -- Driver ([`driver.rs:2793`](https://github.com/warpdotdev/warp/blob/84276e0732860798fa49eb7372e6ee90cdf1728b/app/src/ai/agent_sdk/driver.rs#L2793)): `TransientError` → `end_run_after(AUTO_RESUME_TIMEOUT = 120s)` with the last structured error (I11); a recovery flips status back to `InProgress`, cancelling the deadline. The old `will_attempt_resume` check in the Error arm is gone — `Error` is always terminal now. +- Driver ([`driver.rs:2793`](https://github.com/warpdotdev/warp/blob/84276e0732860798fa49eb7372e6ee90cdf1728b/app/src/ai/agent_sdk/driver.rs#L2793)): `TransientError` → `end_run_after(AUTO_RESUME_TIMEOUT = 120s)` with the last structured error (I11); a recovery flips status back to `InProgress`, cancelling the deadline, and the next failure arms a fresh deadline — so the window is per attempt, not per turn. The old `will_attempt_resume` check in the Error arm is gone — `Error` is always terminal now. - Sync model ([`local_agent_task_sync_model.rs:322`](https://github.com/warpdotdev/warp/blob/84276e0732860798fa49eb7372e6ee90cdf1728b/app/src/ai/blocklist/local_agent_task_sync_model.rs#L322)): `TransientError` → `IN_PROGRESS` with no status message (I6); `task_update_for_conversation_error` ignores the `will_attempt_resume` rendering hint (terminal classification only). - `ambient_agents::conversation_output_status_from_conversation`: early `None` for `TransientError`; for terminal `Error` it now prefers the structured exchange error over `status_error_message`. - Run lists / pill bar / aggregation / notifications / queued-prompt gating treat it as working (I7, I8, I19). @@ -37,11 +37,13 @@ Key files: ### Strict retry/resume split -[`response_stream.rs:41`](https://github.com/warpdotdev/warp/blob/84276e0732860798fa49eb7372e6ee90cdf1728b/app/src/ai/blocklist/controller/response_stream.rs#L41): pure `recovery_action(has_received_client_actions, is_recoverable, has_retry_budget, can_attempt_resume_on_error, is_online) -> {RetryNow, RetryWhenOnline, Resume, Fail}`: +[`response_stream.rs:41`](https://github.com/warpdotdev/warp/blob/84276e0732860798fa49eb7372e6ee90cdf1728b/app/src/ai/blocklist/controller/response_stream.rs#L41): pure `recovery_action(has_received_client_actions, is_recoverable, recovery, is_online) -> {Retry, RetryWhenOnline, Resume, Fail(reason)}`: -- No actions yet + retryable + budget (3) → retry, verbatim re-send (I2). Offline parks the retry (`RetryWhenOnline`). -- Actions received + transient + resume-eligible → one-shot `ResumeConversation` (I3); resumes run with `can_attempt_resume_on_error = false`, bounding recovery (I9). -- Everything else → `Fail` (terminal). Recovery eligibility (retry and resume) uses `AIApiError::is_recoverable()` ([`server_api.rs:341`](https://github.com/warpdotdev/warp/blob/84276e0732860798fa49eb7372e6ee90cdf1728b/app/src/server/server_api.rs#L341)) (I12). +- No actions yet + retryable + budget → retry, verbatim re-send after a backoff (I2). Offline parks the retry instead (`RetryWhenOnline`). +- Actions received + transient + resume-eligible + budget → `ResumeConversation` (I3). +- Everything else → `Fail(reason)` (terminal), where the reason distinguishes a non-transient error, an exhausted budget, and a request that may not resume at all. Recovery eligibility (retry and resume) uses `AIApiError::is_recoverable()` ([`server_api.rs:341`](https://github.com/warpdotdev/warp/blob/84276e0732860798fa49eb7372e6ee90cdf1728b/app/src/server/server_api.rs#L341)) (I12). + +Both kinds of recovery draw from one `RecoveryBudget` (I9), a `Copy` value the request carries: a retry charges it in place, and a resume hands the charged budget to the `ResumeConversation` request the controller sends next, as a `PendingResume` (`pending_resume` → `schedule_auto_resume_after_error`). The scope is one request and its recoveries, not one agent turn — `send_follow_up_for_conversation` and the other outbound paths each start a `RecoveryBudget::fresh()`, as they did when the budget was a per-stream `retry_count`. `resume_allowed = false` now means only "this request kind may never resume" (the passive-request clamp, I17), not "recovery is over". Every attempt waits `retry_strategies::backoff_after_attempts`, the same jittered exponential schedule `with_bounded_retry` uses (I10); the resume's wait is decided with the decision and carried on `PendingResume` rather than recomputed at send time, since the schedule is jittered and a recomputed value would not be the one that was logged. Raised from the original one-shot resume by [REMOTE-2269](https://linear.app/warpdotdev/issue/REMOTE-2269/allow-multiple-resume-attempts). The controller passes `recovery_pending = should_resume_conversation_after_stream_finished()` into the history model; `will_attempt_resume` on `RenderableAIError` remains rendering-only. @@ -55,13 +57,13 @@ The server always sends `StreamFinished`, but a transport cut between chunks rea ### Cancellation and replacement -`cancel_conversation_progress` aborts the parked resume handle and, when no active stream remains, flips a `TransientError` conversation to `Cancelled` directly (I15). `send_request_input` aborts the pending resume for the conversation before sending (I16) and forces `can_attempt_resume_on_error = false` for passive requests (I17). +`cancel_conversation_progress` aborts the parked resume handle and, when no active stream remains, flips a `TransientError` conversation to `Cancelled` directly (I15). `send_request_input` aborts the pending resume for the conversation before sending (I16) and clamps the budget with `RecoveryBudget::without_resume` for passive requests (I17). ## Testing and validation Unit (all in-tree, `cargo nextest run -p warp --lib`): -- `response_stream_tests.rs` — exhaustive `recovery_action` matrix: retry/park/fail pre-actions (I2, I9, I13), resume gating post-actions (I3, I9), budget exhaustion and non-retryable → fail (I10, I12). +- `response_stream_tests.rs` — exhaustive `recovery_action` matrix: retry/park/fail pre-actions (I2, I9, I13), resume gating post-actions (I3, I9), budget exhaustion and non-retryable → fail (I10, I12); the shared budget across a retry/resume mix and across repeated resumes, plus a bound tying the backoff schedule to `AUTO_RESUME_TIMEOUT` (I9, I10, I11). - `server_api_tests.rs` — `is_recoverable` classification: 5xx/timeout/transport and app-level (quota/overload/misc/JSON) recoverable, other 4xx not (I12); `UnexpectedEof` recoverable (I1). - `history_model_tests.rs` — `recovery_pending` → `TransientError` and no terminal derived outcome (I5, I6 upstream); structured exchange error preserved through conversion; non-recoverable error stays terminal. - `local_agent_task_sync_model_tests.rs` — `TransientError` → `IN_PROGRESS` with no status message (I6); `will_attempt_resume` hint ignored for terminal classification (I10, I12). @@ -69,8 +71,8 @@ Unit (all in-tree, `cargo nextest run -p warp --lib`): E2e (oz-local + warp-server `TransportReset` LLM mock, `simulate_maa_transport_reset: true`): - Recovery (I1, I3–I6, I11): single mid-stream reset → client log shows the synthesized truncation → conversation `TransientError` → driver "automatic recovery pending — waiting up to 120s" → resume fires → run completes with the mocked `finish_task`. Verified 2026-06-10. -- Bounded failure (I9, I10): cycling resets → exactly one resume, then terminal ERROR with the friendly lost-connection message; 2 LLM calls total, no retry storm. Verified 2026-06-10. -- The pre-action retry paths (I2) are not reachable through this mock on cloud runs (the user-message-append ClientActions always precede the LLM call), so they are covered by the unit matrix plus the request-send-failure path. +- Bounded failure (I9, I10): cycling resets → exactly one resume, then terminal ERROR with the friendly lost-connection message; 2 LLM calls total, no retry storm. Verified 2026-06-10, against the pre-REMOTE-2269 one-shot resume; the shared budget makes this 3 spaced attempts before the same terminal error. +- The pre-action retry paths (I2) were not exercised by these scenarios, so they are covered by the unit matrix plus the request-send-failure path. The mock cut the stream at the LLM call, by which point the server had already emitted that request's user-message-append `ClientActions` — so it could not produce a failure before the first `ClientActions` event, which is what the retry paths need. That is a limit of where the mock injected the reset, **not** a property of cloud runs: `has_received_client_actions` flips only when a `ClientActions` event actually arrives, so any failure ahead of the first one (initial connection error, pre-header 5xx, early reset) does take the retry path with the full budget. [REMOTE-2269](https://linear.app/warpdotdev/issue/REMOTE-2269/allow-multiple-resume-attempts) records one from a real cloud run: `MultiAgent request failed, retrying (attempt 1/3) - Error: ErrorStatus(503, "upstream connect error or disconnect/reset before headers. reset reason: remote connection failure")`. Lint/build gates: `cargo fmt`, `cargo clippy -p warp --lib --tests -- -D warnings`, `cargo check -p warp --lib --features crash_reporting`. @@ -78,7 +80,7 @@ Not covered by automated tests (manual/review only): controller cancellation-dur ## Risks and mitigations -- **Stuck "Reconnecting"**: any path that sets `TransientError` and never delivers a recovery would hang the UI state. Mitigated by the driver's 120s deadline (cloud), the one-shot resume contract, and the cancellation flip to `Cancelled`; restore-from-disk degrades it to `Error`. +- **Stuck "Reconnecting"**: any path that sets `TransientError` and never delivers a recovery would hang the UI state. Mitigated by the driver's 120s per-attempt deadline (cloud), the shared recovery budget, and the cancellation flip to `Cancelled`; restore-from-disk degrades it to `Error`. - **Behavior change**: quota/overload failures no longer set the resume flag (master did). Intentional — a resume would fail identically — but visible to anyone who relied on the old flag. - **Sentry volume**: synthesized truncations now report (with `will_attempt_resume`/`is_recoverable` tags) — expect new `UnexpectedEof` events that previously appeared as the generic EOF fallback.