From 20de3f49da567c51c11ec87d56f710e8b918ee85 Mon Sep 17 00:00:00 2001 From: seemeroland Date: Wed, 15 Jul 2026 16:50:01 -0700 Subject: [PATCH 1/3] Give up on reconnecting agent event stream on persistent failures --- app/src/ai/agent_events/driver.rs | 116 +++++++++++++++++- app/src/ai/agent_events/driver_tests.rs | 115 +++++++++++++++++ .../harness/claude_code/parent_bridge.rs | 7 +- app/src/server/retry_strategies.rs | 16 +++ 4 files changed, 251 insertions(+), 3 deletions(-) diff --git a/app/src/ai/agent_events/driver.rs b/app/src/ai/agent_events/driver.rs index e918b375afe..1a40da16039 100644 --- a/app/src/ai/agent_events/driver.rs +++ b/app/src/ai/agent_events/driver.rs @@ -9,7 +9,7 @@ use instant::Instant; use warp_errors::{report_error, AnyhowErrorExt as _}; use warpui::r#async::Timer; -use crate::server::retry_strategies::is_transient_http_error; +use crate::server::retry_strategies::{is_auth_error, is_transient_http_error}; use crate::server::server_api::ai::AgentRunEvent; use crate::server::server_api::presigned_upload::HttpStatusError; use crate::server::server_api::ServerApi; @@ -19,6 +19,18 @@ pub(crate) const DEFAULT_PERMANENT_ERROR_BACKOFF_STEPS: &[u64] = &[30]; pub(crate) const DEFAULT_AGENT_EVENT_PROACTIVE_RECONNECT: Duration = Duration::from_secs(14 * 60); pub(crate) const DEFAULT_AGENT_EVENT_FAILURES_BEFORE_ERROR_LOG: usize = 5; +/// Consecutive authentication failures (HTTP 401/403) after which a bounded +/// listener gives up instead of reconnecting. A small threshold (rather than 1) +/// tolerates a one-off token blip while still bailing quickly once credentials +/// are permanently invalid. +pub(crate) const DEFAULT_AUTH_ERROR_GIVE_UP_FAILURES: usize = 3; + +/// Total wall-clock time a bounded listener keeps retrying, measured from the +/// first failure since the last successful open/event, before it gives up. Set +/// larger than the proactive reconnect window so healthy streams are never +/// affected; it only bounds sustained failure. +pub(crate) const DEFAULT_AGENT_EVENT_MAX_RETRY_DURATION: Duration = Duration::from_secs(30 * 60); + /// Selects which server-side filter shape an [`AgentEventSource`] should use /// when opening a stream. /// @@ -82,6 +94,16 @@ pub(crate) struct AgentEventDriverConfig { /// Failure count at which actionable reconnect failures are reported at Error level. /// This only affects log severity; retry behavior stays the same. pub failures_before_error_log: usize, + /// Consecutive authentication failures (HTTP 401/403) after which the driver + /// stops and returns an error instead of reconnecting. `None` keeps retrying + /// through auth errors forever (the default for local, interactive + /// listeners whose credentials refresh in the background). + pub auth_error_give_up_failures: Option, + /// Total wall-clock time the driver keeps retrying after failures begin + /// (measured from the first failure since the last successful open/event) + /// before it stops and returns an error. `None` retries without a time + /// limit. + pub max_retry_duration: Option, } impl AgentEventDriverConfig { @@ -95,6 +117,8 @@ impl AgentEventDriverConfig { permanent_error_backoff_steps: DEFAULT_PERMANENT_ERROR_BACKOFF_STEPS, proactive_reconnect_after: Some(DEFAULT_AGENT_EVENT_PROACTIVE_RECONNECT), failures_before_error_log: DEFAULT_AGENT_EVENT_FAILURES_BEFORE_ERROR_LOG, + auth_error_give_up_failures: None, + max_retry_duration: None, } } @@ -103,6 +127,19 @@ impl AgentEventDriverConfig { pub(crate) fn retry_forever_run_ids(run_ids: Vec, since_sequence: i64) -> Self { Self::retry_forever(AgentEventFilter::RunIds(run_ids), since_sequence) } + + /// Build a reconnecting config for a cloud-agent listener that must NOT run + /// forever. Unlike [`retry_forever`], this stops after sustained + /// authentication failures or a bounded total retry window, so a listener + /// whose task has ended (and whose credentials are now permanently + /// rejected) shuts down instead of reconnecting indefinitely. + pub(crate) fn bounded_run_ids(run_ids: Vec, since_sequence: i64) -> Self { + Self { + auth_error_give_up_failures: Some(DEFAULT_AUTH_ERROR_GIVE_UP_FAILURES), + max_retry_duration: Some(DEFAULT_AGENT_EVENT_MAX_RETRY_DURATION), + ..Self::retry_forever(AgentEventFilter::RunIds(run_ids), since_sequence) + } + } } /// Tells the shared driver whether to continue or stop after a handled event. @@ -300,6 +337,10 @@ where let mut since_sequence = config.since_sequence; let mut failures = 0usize; let mut has_connected_once = false; + // Start of the current run of consecutive failures. Reset to `None` after any + // successful open/event so the `max_retry_duration` window only measures + // sustained failure, not total stream lifetime. + let mut retry_window_started_at: Option = None; loop { // `open_stream` is lazy for the SSE-backed source: the TCP @@ -311,6 +352,17 @@ where Ok(stream) => stream, Err(err) => { failures += 1; + if let Some(reason) = agent_event_give_up_reason( + &config, + failures, + is_auth_error(&err), + &mut retry_window_started_at, + ) { + return Err(err.context(format!( + "Agent event driver {reason} for {}", + config.filter.log_label() + ))); + } let backoff_steps = if is_transient_http_error(&err) { config.reconnect_backoff_steps } else { @@ -368,6 +420,7 @@ where } NextDriverItem::StreamItem(Some(Ok(AgentEventSourceItem::Open))) => { failures = 0; + retry_window_started_at = None; has_connected_once = true; notify_driver_state(consumer, AgentEventDriverState::Connected).await; log::info!( @@ -377,6 +430,7 @@ where } NextDriverItem::StreamItem(Some(Ok(AgentEventSourceItem::Event(event)))) => { failures = 0; + retry_window_started_at = None; if event.sequence <= since_sequence { continue; } @@ -397,6 +451,17 @@ where } NextDriverItem::StreamItem(Some(Err(err))) => { failures += 1; + if let Some(reason) = agent_event_give_up_reason( + &config, + failures, + is_auth_error(&err), + &mut retry_window_started_at, + ) { + return Err(err.context(format!( + "Agent event driver {reason} for {}", + config.filter.log_label() + ))); + } let backoff_steps = if is_transient_http_error(&err) { config.reconnect_backoff_steps } else { @@ -427,6 +492,20 @@ where // there is no HTTP status to classify. NextDriverItem::StreamItem(None) => { failures += 1; + // A clean server-side close carries no HTTP status, so it is + // never treated as an auth failure; only the time-based + // backstop can trigger a give-up here. + if let Some(reason) = agent_event_give_up_reason( + &config, + failures, + false, + &mut retry_window_started_at, + ) { + return Err(anyhow!( + "Agent event driver {reason} for {} after stream closed", + config.filter.log_label() + )); + } let backoff = agent_event_backoff(failures, config.reconnect_backoff_steps); log::warn!( "Agent event stream closed for {}, reconnecting in {backoff:?}", @@ -454,6 +533,41 @@ enum NextDriverItem { ProactiveReconnect, } +/// Decide whether a bounded driver should stop retrying after a failure. +/// +/// Returns `Some(reason)` (a short human-readable explanation for logs) when the +/// driver should give up, or `None` to keep retrying. `retry_window_started_at` +/// is seeded on the first call of a failure run so the `max_retry_duration` +/// window measures sustained failure; callers reset it to `None` on success. +fn agent_event_give_up_reason( + config: &AgentEventDriverConfig, + failures: usize, + is_auth_failure: bool, + retry_window_started_at: &mut Option, +) -> Option { + let now = Instant::now(); + let window_start = *retry_window_started_at.get_or_insert(now); + + if let Some(threshold) = config.auth_error_give_up_failures { + if is_auth_failure && failures >= threshold { + return Some(format!( + "stopping after {failures} consecutive authentication failures" + )); + } + } + + if let Some(max) = config.max_retry_duration { + let elapsed = now.saturating_duration_since(window_start); + if elapsed >= max { + return Some(format!( + "stopping after retrying for {elapsed:?} (max_retry_duration={max:?})" + )); + } + } + + None +} + async fn notify_driver_state( consumer: &mut C, state: AgentEventDriverState, diff --git a/app/src/ai/agent_events/driver_tests.rs b/app/src/ai/agent_events/driver_tests.rs index 2f7f9d337ee..e48cb23cd3e 100644 --- a/app/src/ai/agent_events/driver_tests.rs +++ b/app/src/ai/agent_events/driver_tests.rs @@ -145,6 +145,8 @@ async fn driver_skips_duplicate_sequences_and_persists_new_cursor() { permanent_error_backoff_steps: DEFAULT_PERMANENT_ERROR_BACKOFF_STEPS, proactive_reconnect_after: None, failures_before_error_log: DEFAULT_AGENT_EVENT_FAILURES_BEFORE_ERROR_LOG, + auth_error_give_up_failures: None, + max_retry_duration: None, }; run_agent_event_driver(source, config, &mut consumer) @@ -191,6 +193,8 @@ async fn driver_resets_failures_after_successful_event_delivery() { permanent_error_backoff_steps: DEFAULT_PERMANENT_ERROR_BACKOFF_STEPS, proactive_reconnect_after: None, failures_before_error_log: DEFAULT_AGENT_EVENT_FAILURES_BEFORE_ERROR_LOG, + auth_error_give_up_failures: None, + max_retry_duration: None, }; run_agent_event_driver(source, config, &mut consumer) @@ -236,6 +240,8 @@ async fn driver_ignores_persist_cursor_errors() { permanent_error_backoff_steps: DEFAULT_PERMANENT_ERROR_BACKOFF_STEPS, proactive_reconnect_after: None, failures_before_error_log: DEFAULT_AGENT_EVENT_FAILURES_BEFORE_ERROR_LOG, + auth_error_give_up_failures: None, + max_retry_duration: None, }; run_agent_event_driver(source, config, &mut consumer) @@ -270,6 +276,8 @@ async fn driver_ignores_driver_state_errors() { permanent_error_backoff_steps: DEFAULT_PERMANENT_ERROR_BACKOFF_STEPS, proactive_reconnect_after: None, failures_before_error_log: DEFAULT_AGENT_EVENT_FAILURES_BEFORE_ERROR_LOG, + auth_error_give_up_failures: None, + max_retry_duration: None, }; run_agent_event_driver(source, config, &mut consumer) @@ -306,6 +314,8 @@ async fn driver_retries_initial_connection_until_stream_opens() { permanent_error_backoff_steps: DEFAULT_PERMANENT_ERROR_BACKOFF_STEPS, proactive_reconnect_after: None, failures_before_error_log: DEFAULT_AGENT_EVENT_FAILURES_BEFORE_ERROR_LOG, + auth_error_give_up_failures: None, + max_retry_duration: None, }; run_agent_event_driver(source, config, &mut consumer) @@ -444,6 +454,8 @@ async fn driver_uses_slow_backoff_on_permanent_http_error() { permanent_error_backoff_steps: ZERO_BACKOFF_STEPS, proactive_reconnect_after: None, failures_before_error_log: DEFAULT_AGENT_EVENT_FAILURES_BEFORE_ERROR_LOG, + auth_error_give_up_failures: None, + max_retry_duration: None, }; run_agent_event_driver(source, config, &mut consumer) @@ -463,6 +475,107 @@ async fn driver_uses_slow_backoff_on_permanent_http_error() { assert_eq!(retry_backoff, Duration::from_secs(0)); } +#[tokio::test] +async fn driver_gives_up_after_consecutive_auth_failures() { + // Every open attempt fails with a 401. With a give-up threshold of 3, the + // driver should stop and return an error rather than reconnecting forever. + let source = FakeAgentEventSource::new(vec![ + Err(make_http_status_error(401)), + Err(make_http_status_error(401)), + Err(make_http_status_error(401)), + ]); + let mut consumer = RecordingConsumer { + stop_after: 1, + ..Default::default() + }; + + let config = AgentEventDriverConfig { + filter: AgentEventFilter::RunIds(vec!["child-run".to_string()]), + since_sequence: 0, + reconnect_backoff_steps: ZERO_BACKOFF_STEPS, + permanent_error_backoff_steps: ZERO_BACKOFF_STEPS, + proactive_reconnect_after: None, + failures_before_error_log: DEFAULT_AGENT_EVENT_FAILURES_BEFORE_ERROR_LOG, + auth_error_give_up_failures: Some(3), + max_retry_duration: None, + }; + + let result = run_agent_event_driver(source, config, &mut consumer).await; + assert!( + result.is_err(), + "driver should give up on persistent auth errors" + ); + assert!(consumer.handled_sequences.is_empty()); +} + +#[tokio::test] +async fn driver_does_not_give_up_on_non_auth_error_when_only_auth_bounded() { + // A non-auth (500) error must not trip the auth give-up path: the driver + // reconnects and succeeds. + let source = FakeAgentEventSource::new(vec![ + Err(make_http_status_error(500)), + Err(make_http_status_error(500)), + Err(make_http_status_error(500)), + ok_stream(vec![ + Ok(AgentEventSourceItem::Open), + Ok(AgentEventSourceItem::Event(make_run_event( + 1, + "new_message", + "child-run", + Some("msg-1"), + ))), + ]), + ]); + let mut consumer = RecordingConsumer { + stop_after: 1, + ..Default::default() + }; + + let config = AgentEventDriverConfig { + filter: AgentEventFilter::RunIds(vec!["child-run".to_string()]), + since_sequence: 0, + reconnect_backoff_steps: ZERO_BACKOFF_STEPS, + permanent_error_backoff_steps: ZERO_BACKOFF_STEPS, + proactive_reconnect_after: None, + failures_before_error_log: DEFAULT_AGENT_EVENT_FAILURES_BEFORE_ERROR_LOG, + auth_error_give_up_failures: Some(3), + max_retry_duration: None, + }; + + run_agent_event_driver(source, config, &mut consumer) + .await + .unwrap(); + assert_eq!(consumer.handled_sequences, vec![1]); +} + +#[tokio::test] +async fn driver_gives_up_after_max_retry_duration() { + // A zero-length max retry window means the driver gives up on the first + // failure regardless of error class. + let source = FakeAgentEventSource::new(vec![Err(make_http_status_error(500))]); + let mut consumer = RecordingConsumer { + stop_after: 1, + ..Default::default() + }; + + let config = AgentEventDriverConfig { + filter: AgentEventFilter::RunIds(vec!["child-run".to_string()]), + since_sequence: 0, + reconnect_backoff_steps: ZERO_BACKOFF_STEPS, + permanent_error_backoff_steps: ZERO_BACKOFF_STEPS, + proactive_reconnect_after: None, + failures_before_error_log: DEFAULT_AGENT_EVENT_FAILURES_BEFORE_ERROR_LOG, + auth_error_give_up_failures: None, + max_retry_duration: Some(Duration::from_secs(0)), + }; + + let result = run_agent_event_driver(source, config, &mut consumer).await; + assert!( + result.is_err(), + "driver should give up once the max retry duration elapses" + ); +} + #[tokio::test] async fn driver_uses_fast_backoff_on_transient_http_error() { let source = FakeAgentEventSource::new(vec![ @@ -495,6 +608,8 @@ async fn driver_uses_fast_backoff_on_transient_http_error() { permanent_error_backoff_steps: &[9999], proactive_reconnect_after: None, failures_before_error_log: DEFAULT_AGENT_EVENT_FAILURES_BEFORE_ERROR_LOG, + auth_error_give_up_failures: None, + max_retry_duration: None, }; run_agent_event_driver(source, config, &mut consumer) diff --git a/app/src/ai/agent_sdk/driver/harness/claude_code/parent_bridge.rs b/app/src/ai/agent_sdk/driver/harness/claude_code/parent_bridge.rs index 855b8c59d71..d271d8ef631 100644 --- a/app/src/ai/agent_sdk/driver/harness/claude_code/parent_bridge.rs +++ b/app/src/ai/agent_sdk/driver/harness/claude_code/parent_bridge.rs @@ -688,8 +688,11 @@ async fn run_parent_bridge_forever( // The shared driver keeps `since_sequence` in memory across its own retry // loop and we also persist it inside the session state dir so dormant runs // can resume without replaying already handled events. - let config = - AgentEventDriverConfig::retry_forever_run_ids(vec![run_id.clone()], since_sequence); + // + // Use the bounded config: once this task ends, its credentials are + // permanently rejected (HTTP 401), so the driver should stop rather than + // reconnect forever and keep the worker pod alive indefinitely. + let config = AgentEventDriverConfig::bounded_run_ids(vec![run_id.clone()], since_sequence); let source = ServerApiAgentEventSource::new(server_api.clone()); let mut consumer = MessageBridgeEventConsumer { run_id, diff --git a/app/src/server/retry_strategies.rs b/app/src/server/retry_strategies.rs index 4978ab6cf04..e4ee82d8c36 100644 --- a/app/src/server/retry_strategies.rs +++ b/app/src/server/retry_strategies.rs @@ -91,6 +91,22 @@ fn is_transient_status(status: u16) -> bool { matches!(status, 408 | 429 | 500..=599) } +/// Returns `true` if the error chain carries an [`HttpStatusError`] with an +/// authentication/authorization status (401 or 403). +/// +/// Used by long-lived listeners to distinguish "credentials are permanently +/// invalid" (for example, a cloud-agent task whose token stops working once the +/// task ends) from generic permanent errors, so they can stop retrying instead +/// of reconnecting forever. +pub(crate) fn is_auth_error(e: &anyhow::Error) -> bool { + for cause in e.chain() { + if let Some(http_err) = cause.downcast_ref::() { + return matches!(http_err.status, 401 | 403); + } + } + false +} + /// Maximum total attempts per operation (initial attempt plus retries on transient errors). pub(crate) const MAX_ATTEMPTS: usize = 3; From eec6456827f8d2b461fe1c26ef3b9fb8f54314d0 Mon Sep 17 00:00:00 2001 From: seemeroland Date: Wed, 15 Jul 2026 17:16:47 -0700 Subject: [PATCH 2/3] Count only consecutive auth failures toward auth give-up The auth give-up threshold is documented as consecutive authentication failures, but the give-up check compared it against the generic consecutive-failure counter. A mix such as 500, 500, 401 would therefore stop the driver after a single auth failure. Track a dedicated consecutive_auth_failures counter that increments only on 401/403 and resets on any non-auth failure or success, and check the threshold against it. Add regression tests for the mixed-failure and streak-reset cases. Co-Authored-By: Oz --- app/src/ai/agent_events/driver.rs | 46 +++++++++----- app/src/ai/agent_events/driver_tests.rs | 79 +++++++++++++++++++++++++ 2 files changed, 110 insertions(+), 15 deletions(-) diff --git a/app/src/ai/agent_events/driver.rs b/app/src/ai/agent_events/driver.rs index 1a40da16039..d57b73721d0 100644 --- a/app/src/ai/agent_events/driver.rs +++ b/app/src/ai/agent_events/driver.rs @@ -336,6 +336,11 @@ where { let mut since_sequence = config.since_sequence; let mut failures = 0usize; + // Consecutive authentication failures (HTTP 401/403), tracked separately from + // `failures` so the auth give-up threshold only counts an uninterrupted run + // of auth failures. Any non-auth failure or success resets it, so a mix like + // 500, 500, 401 does not trip a "3 consecutive auth failures" policy. + let mut consecutive_auth_failures = 0usize; let mut has_connected_once = false; // Start of the current run of consecutive failures. Reset to `None` after any // successful open/event so the `max_retry_duration` window only measures @@ -352,10 +357,14 @@ where Ok(stream) => stream, Err(err) => { failures += 1; + if is_auth_error(&err) { + consecutive_auth_failures += 1; + } else { + consecutive_auth_failures = 0; + } if let Some(reason) = agent_event_give_up_reason( &config, - failures, - is_auth_error(&err), + consecutive_auth_failures, &mut retry_window_started_at, ) { return Err(err.context(format!( @@ -420,6 +429,7 @@ where } NextDriverItem::StreamItem(Some(Ok(AgentEventSourceItem::Open))) => { failures = 0; + consecutive_auth_failures = 0; retry_window_started_at = None; has_connected_once = true; notify_driver_state(consumer, AgentEventDriverState::Connected).await; @@ -430,6 +440,7 @@ where } NextDriverItem::StreamItem(Some(Ok(AgentEventSourceItem::Event(event)))) => { failures = 0; + consecutive_auth_failures = 0; retry_window_started_at = None; if event.sequence <= since_sequence { continue; @@ -451,10 +462,14 @@ where } NextDriverItem::StreamItem(Some(Err(err))) => { failures += 1; + if is_auth_error(&err) { + consecutive_auth_failures += 1; + } else { + consecutive_auth_failures = 0; + } if let Some(reason) = agent_event_give_up_reason( &config, - failures, - is_auth_error(&err), + consecutive_auth_failures, &mut retry_window_started_at, ) { return Err(err.context(format!( @@ -493,12 +508,12 @@ where NextDriverItem::StreamItem(None) => { failures += 1; // A clean server-side close carries no HTTP status, so it is - // never treated as an auth failure; only the time-based - // backstop can trigger a give-up here. + // never treated as an auth failure; reset the auth streak and + // let only the time-based backstop trigger a give-up here. + consecutive_auth_failures = 0; if let Some(reason) = agent_event_give_up_reason( &config, - failures, - false, + consecutive_auth_failures, &mut retry_window_started_at, ) { return Err(anyhow!( @@ -536,22 +551,23 @@ enum NextDriverItem { /// Decide whether a bounded driver should stop retrying after a failure. /// /// Returns `Some(reason)` (a short human-readable explanation for logs) when the -/// driver should give up, or `None` to keep retrying. `retry_window_started_at` -/// is seeded on the first call of a failure run so the `max_retry_duration` -/// window measures sustained failure; callers reset it to `None` on success. +/// driver should give up, or `None` to keep retrying. `consecutive_auth_failures` +/// is the length of the current uninterrupted run of HTTP 401/403 failures (the +/// caller resets it on any non-auth failure or success), and +/// `retry_window_started_at` is seeded on the first call of a failure run so the +/// `max_retry_duration` window measures sustained failure. fn agent_event_give_up_reason( config: &AgentEventDriverConfig, - failures: usize, - is_auth_failure: bool, + consecutive_auth_failures: usize, retry_window_started_at: &mut Option, ) -> Option { let now = Instant::now(); let window_start = *retry_window_started_at.get_or_insert(now); if let Some(threshold) = config.auth_error_give_up_failures { - if is_auth_failure && failures >= threshold { + if consecutive_auth_failures >= threshold { return Some(format!( - "stopping after {failures} consecutive authentication failures" + "stopping after {consecutive_auth_failures} consecutive authentication failures" )); } } diff --git a/app/src/ai/agent_events/driver_tests.rs b/app/src/ai/agent_events/driver_tests.rs index e48cb23cd3e..e97c5142258 100644 --- a/app/src/ai/agent_events/driver_tests.rs +++ b/app/src/ai/agent_events/driver_tests.rs @@ -548,6 +548,85 @@ async fn driver_does_not_give_up_on_non_auth_error_when_only_auth_bounded() { assert_eq!(consumer.handled_sequences, vec![1]); } +#[tokio::test] +async fn driver_does_not_count_non_auth_failures_toward_auth_give_up() { + // Regression: the auth give-up must count only *consecutive* auth failures. + // A mix of non-auth failures followed by a single 401 must NOT trip a + // threshold-of-3 policy (only 1 consecutive auth failure has occurred). + let source = FakeAgentEventSource::new(vec![ + Err(make_http_status_error(500)), + Err(make_http_status_error(500)), + Err(make_http_status_error(401)), + ok_stream(vec![ + Ok(AgentEventSourceItem::Open), + Ok(AgentEventSourceItem::Event(make_run_event( + 1, + "new_message", + "child-run", + Some("msg-1"), + ))), + ]), + ]); + let mut consumer = RecordingConsumer { + stop_after: 1, + ..Default::default() + }; + + let config = AgentEventDriverConfig { + filter: AgentEventFilter::RunIds(vec!["child-run".to_string()]), + since_sequence: 0, + reconnect_backoff_steps: ZERO_BACKOFF_STEPS, + permanent_error_backoff_steps: ZERO_BACKOFF_STEPS, + proactive_reconnect_after: None, + failures_before_error_log: DEFAULT_AGENT_EVENT_FAILURES_BEFORE_ERROR_LOG, + auth_error_give_up_failures: Some(3), + max_retry_duration: None, + }; + + run_agent_event_driver(source, config, &mut consumer) + .await + .unwrap(); + assert_eq!(consumer.handled_sequences, vec![1]); +} + +#[tokio::test] +async fn driver_resets_auth_streak_after_non_auth_failure() { + // A non-auth failure in the middle of an auth run resets the streak, so the + // driver only gives up once it sees a *fresh* run of 3 consecutive 401s. + let source = FakeAgentEventSource::new(vec![ + Err(make_http_status_error(401)), + Err(make_http_status_error(500)), + Err(make_http_status_error(401)), + Err(make_http_status_error(401)), + Err(make_http_status_error(401)), + ]); + let mut consumer = RecordingConsumer { + stop_after: 1, + ..Default::default() + }; + + let config = AgentEventDriverConfig { + filter: AgentEventFilter::RunIds(vec!["child-run".to_string()]), + since_sequence: 0, + reconnect_backoff_steps: ZERO_BACKOFF_STEPS, + permanent_error_backoff_steps: ZERO_BACKOFF_STEPS, + proactive_reconnect_after: None, + failures_before_error_log: DEFAULT_AGENT_EVENT_FAILURES_BEFORE_ERROR_LOG, + auth_error_give_up_failures: Some(3), + max_retry_duration: None, + }; + + // If the streak were not reset by the 500, the driver would give up before + // consuming all five responses; a give-up after exactly the last 401 means + // every response was consumed (the fake source panics if over-polled). + let result = run_agent_event_driver(source, config, &mut consumer).await; + assert!( + result.is_err(), + "driver should give up after a fresh run of 3 consecutive auth failures" + ); + assert!(consumer.handled_sequences.is_empty()); +} + #[tokio::test] async fn driver_gives_up_after_max_retry_duration() { // A zero-length max retry window means the driver gives up on the first From 65b8f1727f0dbbce2043cd963a2fda32a0b89f2f Mon Sep 17 00:00:00 2001 From: seemeroland Date: Wed, 15 Jul 2026 17:31:19 -0700 Subject: [PATCH 3/3] Silence wasm dead_code for native-only bounded give-up items The bounded give-up constants and bounded_run_ids are only used by the cloud-agent message bridge, which is not compiled for wasm, so they trip -D dead-code under the wasm clippy target. Gate them with cfg_attr(target_family = "wasm", allow(dead_code)). Co-Authored-By: Oz --- app/src/ai/agent_events/driver.rs | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/app/src/ai/agent_events/driver.rs b/app/src/ai/agent_events/driver.rs index d57b73721d0..1b190ff4406 100644 --- a/app/src/ai/agent_events/driver.rs +++ b/app/src/ai/agent_events/driver.rs @@ -23,12 +23,16 @@ pub(crate) const DEFAULT_AGENT_EVENT_FAILURES_BEFORE_ERROR_LOG: usize = 5; /// listener gives up instead of reconnecting. A small threshold (rather than 1) /// tolerates a one-off token blip while still bailing quickly once credentials /// are permanently invalid. +// Only consumed by the (native-only) cloud-agent message bridge via +// `bounded_run_ids`, so it is dead code on the wasm target. +#[cfg_attr(target_family = "wasm", allow(dead_code))] pub(crate) const DEFAULT_AUTH_ERROR_GIVE_UP_FAILURES: usize = 3; /// Total wall-clock time a bounded listener keeps retrying, measured from the /// first failure since the last successful open/event, before it gives up. Set /// larger than the proactive reconnect window so healthy streams are never /// affected; it only bounds sustained failure. +#[cfg_attr(target_family = "wasm", allow(dead_code))] pub(crate) const DEFAULT_AGENT_EVENT_MAX_RETRY_DURATION: Duration = Duration::from_secs(30 * 60); /// Selects which server-side filter shape an [`AgentEventSource`] should use @@ -133,6 +137,9 @@ impl AgentEventDriverConfig { /// authentication failures or a bounded total retry window, so a listener /// whose task has ended (and whose credentials are now permanently /// rejected) shuts down instead of reconnecting indefinitely. + // Only called by the (native-only) cloud-agent message bridge, so it is dead + // code on the wasm target. + #[cfg_attr(target_family = "wasm", allow(dead_code))] pub(crate) fn bounded_run_ids(run_ids: Vec, since_sequence: i64) -> Self { Self { auth_error_give_up_failures: Some(DEFAULT_AUTH_ERROR_GIVE_UP_FAILURES),