From aae7976054b248ef6133002b4de4c90464904346 Mon Sep 17 00:00:00 2001 From: obchain Date: Thu, 14 May 2026 12:42:18 +0530 Subject: [PATCH 1/2] fix(composio): retry once on post-OAuth auth-error gap (#1688) Composio reports `connection.status == ACTIVE` ~1-2s after OAuth, but the action-execution gateway takes another 30-60s to sync the token, returning the literal "Connection error, try to authenticate" on the first call. Add a single-shot retry after 8s in the shared composio execute path so the orchestrator no longer surfaces a misleading "not connected" message on the very first turn after OAuth. The retry is gated on a small constant list so a real revoked or mis-scoped connection still surfaces after exactly one round-trip. Transport-level errors keep their existing classification upstream. Refs #1688 --- src/openhuman/composio/action_tool.rs | 7 +- src/openhuman/composio/auth_retry.rs | 90 ++++++++ src/openhuman/composio/auth_retry_tests.rs | 245 +++++++++++++++++++++ src/openhuman/composio/mod.rs | 1 + src/openhuman/composio/tools.rs | 2 +- 5 files changed, 340 insertions(+), 5 deletions(-) create mode 100644 src/openhuman/composio/auth_retry.rs create mode 100644 src/openhuman/composio/auth_retry_tests.rs diff --git a/src/openhuman/composio/action_tool.rs b/src/openhuman/composio/action_tool.rs index b22eb824652..e3e6947b588 100644 --- a/src/openhuman/composio/action_tool.rs +++ b/src/openhuman/composio/action_tool.rs @@ -117,10 +117,9 @@ impl Tool for ComposioActionTool { } let started = std::time::Instant::now(); - let res = self - .client - .execute_tool(&self.action_name, Some(args)) - .await; + let res = + super::auth_retry::execute_with_auth_retry(&self.client, &self.action_name, Some(args)) + .await; let elapsed_ms = started.elapsed().as_millis() as u64; match res { diff --git a/src/openhuman/composio/auth_retry.rs b/src/openhuman/composio/auth_retry.rs new file mode 100644 index 00000000000..cb0ebe9fb23 --- /dev/null +++ b/src/openhuman/composio/auth_retry.rs @@ -0,0 +1,90 @@ +//! Single-shot retry wrapper around [`ComposioClient::execute_tool`] for +//! the post-OAuth token-propagation gap (issue #1688). +//! +//! Composio reports `connection.status == ACTIVE` ~1-2s after the user +//! finishes OAuth, but its action-execution gateway can take another +//! 30-60s to sync the new token into its execution cache, run scope +//! validation against the upstream provider, and step out of its +//! first-use rate limit. During that window the gateway returns the +//! literal string `"Connection error, try to authenticate"` for normal +//! action calls — the connection is genuinely active and a second call +//! seconds later succeeds. +//! +//! The retry here is intentionally narrow: +//! +//! * single re-attempt only, after a fixed short sleep +//! * gated on a small constant list of well-known auth-error strings so +//! a real revoked or mis-scoped connection still surfaces to the user +//! after exactly one round-trip — never an infinite loop +//! * only the payload-level `successful = false / error = "…"` shape is +//! eligible; transport-level errors (HTTP non-2xx, bad envelope, connect +//! failures) propagate unchanged because the upstream +//! [`crate::openhuman::integrations`] client already classifies and +//! retries those separately. + +use std::time::Duration; + +use super::client::ComposioClient; +use super::types::ComposioExecuteResponse; + +/// Literal error strings Composio's gateway emits during the post-OAuth +/// readiness gap. Matching is `error.contains(needle)` so trailing +/// punctuation and capitalisation drift on the gateway side does not +/// silently disable the retry. +pub(crate) const RETRYABLE_AUTH_ERRORS: &[&str] = &["Connection error, try to authenticate"]; + +/// Backoff before the single retry. 8s sits in the middle of the 5-10s +/// recommendation in issue #1688 — long enough for Composio's action +/// gateway to sync the token, short enough that a genuine auth failure +/// surfaces to the user well inside the orchestrator's per-turn budget. +pub(crate) const AUTH_RETRY_BACKOFF: Duration = Duration::from_secs(8); + +/// Execute `slug` against the Composio gateway and, on a known +/// post-OAuth auth-error payload, retry exactly once after +/// [`AUTH_RETRY_BACKOFF`]. The second response is returned verbatim, +/// even if it is itself an error — callers see exactly what the gateway +/// produced. +pub(crate) async fn execute_with_auth_retry( + client: &ComposioClient, + slug: &str, + args: Option, +) -> anyhow::Result { + execute_with_auth_retry_inner(client, slug, args, AUTH_RETRY_BACKOFF).await +} + +/// Test-visible inner form that takes an explicit backoff so unit tests +/// can drive the retry path without sleeping for real seconds. +pub(crate) async fn execute_with_auth_retry_inner( + client: &ComposioClient, + slug: &str, + args: Option, + backoff: Duration, +) -> anyhow::Result { + let first = client.execute_tool(slug, args.clone()).await?; + if first.successful { + return Ok(first); + } + let err_text = first.error.as_deref().unwrap_or(""); + if err_text.is_empty() || !is_retryable_auth_error(err_text) { + return Ok(first); + } + tracing::warn!( + target: "composio", + slug = %slug, + error = %err_text, + sleep_ms = backoff.as_millis() as u64, + "[composio] post-OAuth auth error on first action call; sleeping and retrying once (#1688)" + ); + tokio::time::sleep(backoff).await; + client.execute_tool(slug, args).await +} + +fn is_retryable_auth_error(err: &str) -> bool { + RETRYABLE_AUTH_ERRORS + .iter() + .any(|needle| err.contains(needle)) +} + +#[cfg(test)] +#[path = "auth_retry_tests.rs"] +mod tests; diff --git a/src/openhuman/composio/auth_retry_tests.rs b/src/openhuman/composio/auth_retry_tests.rs new file mode 100644 index 00000000000..ca44eecf038 --- /dev/null +++ b/src/openhuman/composio/auth_retry_tests.rs @@ -0,0 +1,245 @@ +//! Tests for the post-OAuth auth-error retry in [`super`]. +//! +//! These spin up a tiny axum backend that mimics Composio's +//! `/agent-integrations/composio/execute` route. Each test wires a +//! response sequence keyed by the request counter so we can assert +//! exactly how many times the gateway was hit. The backoff between +//! attempts is passed in as `Duration::from_millis(0)` so the suite +//! never sleeps for real seconds. + +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +use axum::{routing::post, Json, Router}; +use serde_json::{json, Value}; + +use super::*; +use crate::openhuman::integrations::IntegrationClient; + +async fn start_mock_backend(app: Router) -> String { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + format!("http://127.0.0.1:{}", addr.port()) +} + +fn build_client_for(base_url: String) -> ComposioClient { + let inner = Arc::new(IntegrationClient::new(base_url, "test-token".into())); + ComposioClient::new(inner) +} + +/// First call returns the post-OAuth auth-error payload; second call +/// returns a normal success. Helper must hit the backend twice and +/// surface the second response. +#[tokio::test] +async fn retries_once_on_post_oauth_auth_error_then_succeeds() { + let counter = Arc::new(AtomicUsize::new(0)); + let counter_handler = counter.clone(); + let app = Router::new().route( + "/agent-integrations/composio/execute", + post(move |Json(_body): Json| { + let counter = counter_handler.clone(); + async move { + let n = counter.fetch_add(1, Ordering::SeqCst); + if n == 0 { + Json(json!({ + "success": true, + "data": { + "data": {}, + "successful": false, + "error": "Connection error, try to authenticate", + "costUsd": 0.0 + } + })) + } else { + Json(json!({ + "success": true, + "data": { + "data": { "ok": true }, + "successful": true, + "error": null, + "costUsd": 0.0012 + } + })) + } + } + }), + ); + let base = start_mock_backend(app).await; + let client = build_client_for(base); + + let resp = execute_with_auth_retry_inner( + &client, + "GOOGLECALENDAR_EVENTS_LIST", + Some(json!({})), + Duration::from_millis(0), + ) + .await + .expect("retry path must surface a response"); + + assert!(resp.successful, "second attempt should report success"); + assert_eq!(resp.data["ok"], true); + assert_eq!( + counter.load(Ordering::SeqCst), + 2, + "gateway should be hit exactly twice" + ); +} + +/// A real authentication failure (revoked token, mis-scoped connection, +/// …) returns a 401-equivalent payload that does **not** match the +/// post-OAuth gap string. The helper must surface it after exactly one +/// attempt so the user sees the error without a needless 8s wait. +#[tokio::test] +async fn does_not_retry_on_unrelated_error_payload() { + let counter = Arc::new(AtomicUsize::new(0)); + let counter_handler = counter.clone(); + let app = Router::new().route( + "/agent-integrations/composio/execute", + post(move |Json(_body): Json| { + let counter = counter_handler.clone(); + async move { + counter.fetch_add(1, Ordering::SeqCst); + Json(json!({ + "success": true, + "data": { + "data": {}, + "successful": false, + "error": "invalid_grant: refresh token revoked", + "costUsd": 0.0 + } + })) + } + }), + ); + let base = start_mock_backend(app).await; + let client = build_client_for(base); + + let resp = execute_with_auth_retry_inner( + &client, + "GMAIL_SEND_EMAIL", + Some(json!({"to": "a@b.com"})), + Duration::from_millis(0), + ) + .await + .expect("non-retryable payload must still resolve cleanly"); + + assert!(!resp.successful); + assert_eq!( + resp.error.as_deref(), + Some("invalid_grant: refresh token revoked") + ); + assert_eq!( + counter.load(Ordering::SeqCst), + 1, + "non-retryable errors must not trigger a second attempt" + ); +} + +/// Successful first attempt must short-circuit before the sleep — no +/// retry, no wasted round-trip. +#[tokio::test] +async fn does_not_retry_on_first_attempt_success() { + let counter = Arc::new(AtomicUsize::new(0)); + let counter_handler = counter.clone(); + let app = Router::new().route( + "/agent-integrations/composio/execute", + post(move |Json(_body): Json| { + let counter = counter_handler.clone(); + async move { + counter.fetch_add(1, Ordering::SeqCst); + Json(json!({ + "success": true, + "data": { + "data": { "echoed": true }, + "successful": true, + "error": null, + "costUsd": 0.0 + } + })) + } + }), + ); + let base = start_mock_backend(app).await; + let client = build_client_for(base); + + let resp = execute_with_auth_retry_inner( + &client, + "GITHUB_USERS_GET_AUTHENTICATED", + None, + Duration::from_secs(60), // would hang the test if we ever slept + ) + .await + .unwrap(); + + assert!(resp.successful); + assert_eq!(counter.load(Ordering::SeqCst), 1); +} + +/// If Composio still returns the auth-error payload on the second call +/// (gateway not actually recovered, or real credential problem +/// masquerading as the post-OAuth string), surface the second response +/// verbatim — exactly one retry, never a loop. +#[tokio::test] +async fn retries_once_only_even_when_second_call_still_errors() { + let counter = Arc::new(AtomicUsize::new(0)); + let counter_handler = counter.clone(); + let app = Router::new().route( + "/agent-integrations/composio/execute", + post(move |Json(_body): Json| { + let counter = counter_handler.clone(); + async move { + counter.fetch_add(1, Ordering::SeqCst); + Json(json!({ + "success": true, + "data": { + "data": {}, + "successful": false, + "error": "Connection error, try to authenticate", + "costUsd": 0.0 + } + })) + } + }), + ); + let base = start_mock_backend(app).await; + let client = build_client_for(base); + + let resp = + execute_with_auth_retry_inner(&client, "NOTION_PAGES_LIST", None, Duration::from_millis(0)) + .await + .unwrap(); + + assert!(!resp.successful); + assert_eq!( + resp.error.as_deref(), + Some("Connection error, try to authenticate") + ); + assert_eq!( + counter.load(Ordering::SeqCst), + 2, + "must retry exactly once, never a third time" + ); +} + +#[test] +fn is_retryable_auth_error_matches_known_string() { + assert!(super::is_retryable_auth_error( + "Connection error, try to authenticate" + )); + // Tolerates surrounding whitespace / wrapping text — Composio + // sometimes wraps the message in a longer envelope. + assert!(super::is_retryable_auth_error( + "Action failed: Connection error, try to authenticate (gateway code 401)" + )); +} + +#[test] +fn is_retryable_auth_error_rejects_unrelated_messages() { + assert!(!super::is_retryable_auth_error("invalid_grant")); + assert!(!super::is_retryable_auth_error("ratelimited")); + assert!(!super::is_retryable_auth_error("")); +} diff --git a/src/openhuman/composio/mod.rs b/src/openhuman/composio/mod.rs index 49b52e477c1..03c55f46a84 100644 --- a/src/openhuman/composio/mod.rs +++ b/src/openhuman/composio/mod.rs @@ -36,6 +36,7 @@ //! crate::core::event_bus::DomainEvent::ComposioTriggerReceived pub mod action_tool; +pub mod auth_retry; pub mod bus; pub mod client; pub mod ops; diff --git a/src/openhuman/composio/tools.rs b/src/openhuman/composio/tools.rs index 4f5a65dc08c..ac6aceb2cdc 100644 --- a/src/openhuman/composio/tools.rs +++ b/src/openhuman/composio/tools.rs @@ -697,7 +697,7 @@ impl Tool for ComposioExecuteTool { } let started = std::time::Instant::now(); - let res = self.client.execute_tool(&tool, arguments).await; + let res = super::auth_retry::execute_with_auth_retry(&self.client, &tool, arguments).await; let elapsed_ms = started.elapsed().as_millis() as u64; match res { Ok(mut resp) => { From c2f049e6111bceb6e17ba8d7564a517bdd9df0bb Mon Sep 17 00:00:00 2001 From: obchain Date: Thu, 14 May 2026 13:01:46 +0530 Subject: [PATCH 2/2] fix(composio): redact auth-retry warn log + case-insensitive matcher MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address CodeRabbit on PR #1708: - Drop raw `err_text` from the warn line and log a static `retry_reason` label instead. Provider error strings can embed identifiers (emails, channel/file IDs) and a warn at every retry would broadcast them. - Make the auth-error matcher case-insensitive — the doc comment said "tolerates capitalisation drift" but the implementation used plain case-sensitive `contains`. - Add debug-level breadcrumbs on the start / first-success / non-retry / post-retry branches for observability parity with the rest of the composio module. - Extend the matcher tests with mixed-case fixtures. --- src/openhuman/composio/auth_retry.rs | 55 +++++++++++++++++++--- src/openhuman/composio/auth_retry_tests.rs | 11 ++++- 2 files changed, 58 insertions(+), 8 deletions(-) diff --git a/src/openhuman/composio/auth_retry.rs b/src/openhuman/composio/auth_retry.rs index cb0ebe9fb23..3681d6e3b4a 100644 --- a/src/openhuman/composio/auth_retry.rs +++ b/src/openhuman/composio/auth_retry.rs @@ -60,29 +60,72 @@ pub(crate) async fn execute_with_auth_retry_inner( args: Option, backoff: Duration, ) -> anyhow::Result { + tracing::debug!( + target: "composio", + slug = %slug, + has_args = args.is_some(), + "[composio][auth_retry] execute start" + ); let first = client.execute_tool(slug, args.clone()).await?; if first.successful { + tracing::debug!( + target: "composio", + slug = %slug, + "[composio][auth_retry] first attempt successful; no retry" + ); return Ok(first); } let err_text = first.error.as_deref().unwrap_or(""); - if err_text.is_empty() || !is_retryable_auth_error(err_text) { + let matched = match_retryable_auth_error(err_text); + let Some(matched) = matched else { + // Surface the response unchanged. We deliberately do NOT log + // `err_text` here — upstream provider messages can embed + // identifiers (emails, file IDs, channel names) and a `warn` + // line at every non-retryable failure would broadcast them. + tracing::debug!( + target: "composio", + slug = %slug, + has_error = !err_text.is_empty(), + "[composio][auth_retry] non-retryable payload; returning first response" + ); return Ok(first); - } + }; tracing::warn!( target: "composio", slug = %slug, - error = %err_text, + retry_reason = matched, sleep_ms = backoff.as_millis() as u64, "[composio] post-OAuth auth error on first action call; sleeping and retrying once (#1688)" ); tokio::time::sleep(backoff).await; - client.execute_tool(slug, args).await + let second = client.execute_tool(slug, args).await?; + tracing::debug!( + target: "composio", + slug = %slug, + successful = second.successful, + "[composio][auth_retry] retry attempt completed" + ); + Ok(second) } -fn is_retryable_auth_error(err: &str) -> bool { +/// Returns the matched needle (a static label safe to log) when the +/// provider error text matches one of [`RETRYABLE_AUTH_ERRORS`]. Match +/// is case-insensitive so capitalisation drift on Composio's side does +/// not silently disable the retry. +fn match_retryable_auth_error(err: &str) -> Option<&'static str> { + if err.is_empty() { + return None; + } + let err_lc = err.to_ascii_lowercase(); RETRYABLE_AUTH_ERRORS .iter() - .any(|needle| err.contains(needle)) + .copied() + .find(|needle| err_lc.contains(&needle.to_ascii_lowercase())) +} + +#[cfg(test)] +fn is_retryable_auth_error(err: &str) -> bool { + match_retryable_auth_error(err).is_some() } #[cfg(test)] diff --git a/src/openhuman/composio/auth_retry_tests.rs b/src/openhuman/composio/auth_retry_tests.rs index ca44eecf038..a043c332fdf 100644 --- a/src/openhuman/composio/auth_retry_tests.rs +++ b/src/openhuman/composio/auth_retry_tests.rs @@ -230,11 +230,18 @@ fn is_retryable_auth_error_matches_known_string() { assert!(super::is_retryable_auth_error( "Connection error, try to authenticate" )); - // Tolerates surrounding whitespace / wrapping text — Composio - // sometimes wraps the message in a longer envelope. + // Tolerates wrapping text — Composio sometimes wraps the message + // in a longer envelope. assert!(super::is_retryable_auth_error( "Action failed: Connection error, try to authenticate (gateway code 401)" )); + // Tolerates capitalisation drift on the gateway side. + assert!(super::is_retryable_auth_error( + "CONNECTION ERROR, TRY TO AUTHENTICATE" + )); + assert!(super::is_retryable_auth_error( + "connection error, try to authenticate" + )); } #[test]