diff --git a/app/src/components/composio/ComposioConnectModal.test.tsx b/app/src/components/composio/ComposioConnectModal.test.tsx index a29599951..153fa798b 100644 --- a/app/src/components/composio/ComposioConnectModal.test.tsx +++ b/app/src/components/composio/ComposioConnectModal.test.tsx @@ -446,6 +446,22 @@ describe(' — needs-subdomain recovery phase', () => { }); }); + it('surfaces Meta rate-limit guidance for Instagram authorize failures', async () => { + const instagramToolkit = composioToolkitMeta('instagram'); + vi.mocked(authorize).mockRejectedValueOnce( + new Error('Authorization failed: Backend returned 429 Too Many Requests') + ); + + render( {}} />); + fireEvent.click(screen.getByRole('button', { name: /Connect Instagram/i })); + + await waitFor(() => { + expect(screen.getByText(/Business or Creator account/i)).toBeInTheDocument(); + expect(screen.getByText(/HTTP 429/i)).toBeInTheDocument(); + expect(screen.queryByText(/api.tinyhumans.ai/i)).not.toBeInTheDocument(); + }); + }); + it('surfaces a sanitized (non-raw) error for unrelated authorization failures', async () => { vi.mocked(authorize).mockRejectedValueOnce( new Error( diff --git a/app/src/components/composio/ComposioConnectModal.tsx b/app/src/components/composio/ComposioConnectModal.tsx index e1cfe7e58..6aba375df 100644 --- a/app/src/components/composio/ComposioConnectModal.tsx +++ b/app/src/components/composio/ComposioConnectModal.tsx @@ -33,6 +33,11 @@ import { listConnections, setUserScopes, } from '../../lib/composio/composioApi'; +import { + isMetaOAuthToolkit, + isOAuthRateLimitedError, + metaOAuthRateLimitMessage, +} from '../../lib/composio/oauthHandoff'; import { type ComposioConnection, type ComposioUserScopePref, @@ -172,6 +177,8 @@ export default function ComposioConnectModal({ const pollDeadlineRef = useRef(0); const isPollingRef = useRef(false); const inFlightRef = useRef(false); + const connectInFlightRef = useRef(false); + const [connectInFlight, setConnectInFlight] = useState(false); const initialState = deriveComposioState(connection); const initiallyConnected = initialState === 'connected'; @@ -300,7 +307,7 @@ export default function ComposioConnectModal({ // Fire once immediately, then recurse via setTimeout once the previous // tick resolves. Avoids overlapping async ticks entirely. void tick(); - }, [onChanged, stopPolling, toolkit.slug]); + }, [onChanged, stopPolling, t, toolkit.slug]); // If the modal opens while an OAuth handoff is already in flight // (status = PENDING/INITIATED/…), resume polling instead of asking @@ -327,8 +334,17 @@ export default function ComposioConnectModal({ }, [requiredFields, fieldValues]); const handleConnect = useCallback(async () => { + if (connectInFlightRef.current) { + console.debug( + '[composio][authorize] ignored duplicate Connect click toolkit=%s', + toolkit.slug + ); + return; + } if (!validateRequiredFields()) return; + connectInFlightRef.current = true; + setConnectInFlight(true); setPhase('authorizing'); setError(null); setFieldErrors({}); @@ -392,9 +408,24 @@ export default function ComposioConnectModal({ } setPhase('error'); - setError(sanitizeAuthError(err)); + if (isMetaOAuthToolkit(toolkit.slug) && isOAuthRateLimitedError(err)) { + setError(metaOAuthRateLimitMessage(toolkit.name)); + } else { + setError(sanitizeAuthError(err)); + } + } finally { + connectInFlightRef.current = false; + setConnectInFlight(false); } - }, [validateRequiredFields, requiredFields, fieldValues, startPolling, toolkit.slug]); + }, [ + validateRequiredFields, + requiredFields, + fieldValues, + startPolling, + toolkit.slug, + toolkit.name, + t, + ]); // Fetch the stored scope pref whenever the modal lands in the // 'connected' phase. Re-fetching each time we transition (rather @@ -417,7 +448,7 @@ export default function ComposioConnectModal({ return () => { cancelled = true; }; - }, [phase, toolkit.slug]); + }, [phase, t, toolkit.slug]); const handleToggleScope = useCallback( async (key: keyof ComposioUserScopePref) => { @@ -465,7 +496,7 @@ export default function ComposioConnectModal({ setSavingScope(null); } }, - [savingScope, scopes, toolkit.slug] + [savingScope, scopes, t, toolkit.slug] ); const handleDisconnect = useCallback(async () => { @@ -482,7 +513,7 @@ export default function ComposioConnectModal({ setPhase('error'); setError(`${t('composio.connect.disconnectFailed')}: ${msg}`); } - }, [activeConnection, onChanged]); + }, [activeConnection, onChanged, t]); const handleBackdropClick = (e: React.MouseEvent) => { if (e.target === e.currentTarget) onClose(); @@ -577,8 +608,9 @@ export default function ComposioConnectModal({ {error && phase === 'idle' &&

{error}

} @@ -607,8 +639,9 @@ export default function ComposioConnectModal({ /> diff --git a/app/src/components/composio/toolkitMeta.test.tsx b/app/src/components/composio/toolkitMeta.test.tsx index 2129baca7..b0ff99c24 100644 --- a/app/src/components/composio/toolkitMeta.test.tsx +++ b/app/src/components/composio/toolkitMeta.test.tsx @@ -24,6 +24,12 @@ describe('composioToolkitMeta', () => { expect(calendar.logoUrl).toContain('/googlecalendar'); }); + it('documents Instagram Business account requirement and Meta 429 guidance', () => { + const meta = composioToolkitMeta('instagram'); + expect(meta.description).toMatch(/Business or Creator/i); + expect(meta.description).toMatch(/429/i); + }); + it('falls back cleanly for unknown slugs', () => { const meta = composioToolkitMeta('my_custom_toolkit'); diff --git a/app/src/components/composio/toolkitMeta.tsx b/app/src/components/composio/toolkitMeta.tsx index a757666b1..a4b4913c7 100644 --- a/app/src/components/composio/toolkitMeta.tsx +++ b/app/src/components/composio/toolkitMeta.tsx @@ -322,6 +322,16 @@ export const KNOWN_COMPOSIO_TOOLKITS = Object.freeze( MANAGED_COMPOSIO_TOOLKITS.map(entry => entry.slug) ); +function descriptionForToolkit(key: string, name: string, category: SkillCategory): string { + if (key === 'instagram') { + return ( + 'Connect Instagram Business or Creator accounts (personal accounts are not supported). ' + + 'If Meta shows “Too Many Requests” (HTTP 429), wait a few minutes before retrying.' + ); + } + return defaultDescription(name, category); +} + export function composioToolkitMeta(slug: string): ComposioToolkitMeta { const key = canonicalizeComposioToolkitSlug(slug); const name = MANAGED_TOOLKIT_NAME_BY_SLUG.get(key) ?? prettifyUnknownSlug(key); @@ -329,7 +339,7 @@ export function composioToolkitMeta(slug: string): ComposioToolkitMeta { return { slug: key, name, - description: defaultDescription(name, category), + description: descriptionForToolkit(key, name, category), category, icon: , logoUrl: composioLogoUrl(key), diff --git a/app/src/lib/composio/oauthHandoff.test.ts b/app/src/lib/composio/oauthHandoff.test.ts new file mode 100644 index 000000000..99dbcca27 --- /dev/null +++ b/app/src/lib/composio/oauthHandoff.test.ts @@ -0,0 +1,36 @@ +import { describe, expect, it } from 'vitest'; + +import { + isMetaOAuthToolkit, + isOAuthRateLimitedError, + metaOAuthRateLimitMessage, +} from './oauthHandoff'; + +describe('oauthHandoff', () => { + it('detects Meta OAuth toolkits', () => { + expect(isMetaOAuthToolkit('instagram')).toBe(true); + expect(isMetaOAuthToolkit('Facebook')).toBe(true); + expect(isMetaOAuthToolkit('gmail')).toBe(false); + }); + + it('detects OAuth rate-limit errors', () => { + expect(isOAuthRateLimitedError(new Error('HTTP 429 Too Many Requests'))).toBe(true); + expect(isOAuthRateLimitedError({ message: 'HTTP 429 Too Many Requests' })).toBe(true); + expect(isOAuthRateLimitedError(new Error('rate_limit exceeded'))).toBe(true); + expect(isOAuthRateLimitedError(new Error('401 Unauthorized'))).toBe(false); + }); + + it('builds Instagram-specific Meta rate-limit guidance', () => { + const msg = metaOAuthRateLimitMessage('Instagram'); + expect(msg).toContain('429'); + expect(msg.toLowerCase()).toContain('business'); + }); + + it('builds Facebook-specific Meta rate-limit guidance without Instagram account copy', () => { + const msg = metaOAuthRateLimitMessage('Facebook'); + expect(msg).toContain('429'); + expect(msg).toContain('Facebook'); + expect(msg).toContain('Business Manager'); + expect(msg).not.toContain('Instagram Business or Creator'); + }); +}); diff --git a/app/src/lib/composio/oauthHandoff.ts b/app/src/lib/composio/oauthHandoff.ts new file mode 100644 index 000000000..0c5081008 --- /dev/null +++ b/app/src/lib/composio/oauthHandoff.ts @@ -0,0 +1,50 @@ +/** + * OAuth handoff helpers for Meta-owned Composio toolkits (#1952). + * + * Instagram and Facebook share Meta's OAuth rate limits. The UI uses these + * helpers to detect rate-limit failures and avoid duplicate authorize calls. + */ + +/** Toolkits whose OAuth flows are hosted by Meta. */ +export const META_OAUTH_TOOLKITS = ['instagram', 'facebook'] as const; + +export type MetaOAuthToolkit = (typeof META_OAUTH_TOOLKITS)[number]; + +export function isMetaOAuthToolkit(slug: string): slug is MetaOAuthToolkit { + const key = slug.trim().toLowerCase(); + return (META_OAUTH_TOOLKITS as readonly string[]).includes(key); +} + +/** True when an error message looks like an OAuth / Meta rate limit (HTTP 429). */ +export function isOAuthRateLimitedError(err: unknown): boolean { + if (!err) return false; + const msg = + err instanceof Error + ? err.message + : typeof err === 'object' && err !== null && 'message' in err + ? String((err as { message?: unknown }).message ?? '') + : String(err); + const lower = msg.toLowerCase(); + return ( + lower.includes('429') || + lower.includes('too many requests') || + lower.includes('rate limit') || + lower.includes('rate_limit') || + lower.includes('ratelimited') + ); +} + +/** User-facing copy when Meta OAuth is rate-limited. */ +export function metaOAuthRateLimitMessage(toolkitName: string): string { + const normalizedName = toolkitName.trim().toLowerCase(); + const accountHint = + normalizedName === 'instagram' + ? ' Use an Instagram Business or Creator account — personal accounts are not supported.' + : normalizedName === 'facebook' + ? ' Confirm the Facebook account has access to the relevant Page or Business Manager.' + : ''; + return ( + `Meta is temporarily rate-limiting ${toolkitName} sign-in (HTTP 429). ` + + `Wait a few minutes before retrying and avoid clicking Connect repeatedly.${accountHint}` + ); +} diff --git a/src/openhuman/composio/mod.rs b/src/openhuman/composio/mod.rs index 8851ab9e5..9810fecb2 100644 --- a/src/openhuman/composio/mod.rs +++ b/src/openhuman/composio/mod.rs @@ -43,6 +43,7 @@ pub mod error_mapping; pub mod execute_dispatch; pub mod execute_prepare; pub mod googlecalendar_args; +pub mod oauth_handoff; pub mod ops; pub mod periodic; pub mod providers; diff --git a/src/openhuman/composio/oauth_handoff.rs b/src/openhuman/composio/oauth_handoff.rs new file mode 100644 index 000000000..d0681133f --- /dev/null +++ b/src/openhuman/composio/oauth_handoff.rs @@ -0,0 +1,190 @@ +//! OAuth handoff helpers — Meta (Instagram / Facebook) rate-limit mitigations (#1952). +//! +//! Meta's OAuth authorize endpoint returns HTTP 429 when too many OAuth sessions +//! are created in a short window. That often happens when a user retries after a +//! failed handoff or clicks Connect multiple times, leaving several `PENDING` +//! Composio rows that each redirect through Meta. Before starting a new handoff +//! for Meta-owned toolkits we clear prior non-active connection rows and apply +//! a small backoff retry when the backend reports a 429-shaped failure. + +use std::time::Duration; + +use super::client::{direct_authorize, ComposioClient}; +use super::types::ComposioAuthorizeResponse; + +/// Toolkits whose OAuth flows are hosted by Meta and share the same rate limits. +pub const META_OAUTH_TOOLKITS: &[&str] = &["instagram", "facebook"]; + +const AUTHORIZE_RATE_LIMIT_MAX_ATTEMPTS: u32 = 3; +const AUTHORIZE_RATE_LIMIT_INITIAL_BACKOFF: Duration = Duration::from_secs(5); +const AUTHORIZE_RATE_LIMIT_MAX_BACKOFF: Duration = Duration::from_secs(60); + +/// Return true when `toolkit` uses Meta-hosted OAuth (Instagram / Facebook). +pub fn is_meta_oauth_toolkit(toolkit: &str) -> bool { + let key = toolkit.trim().to_ascii_lowercase(); + META_OAUTH_TOOLKITS.contains(&key.as_str()) +} + +/// Status values that mean an OAuth handoff is still in flight. +pub fn is_inflight_oauth_status(status: &str) -> bool { + matches!( + status.trim().to_ascii_uppercase().as_str(), + "PENDING" | "INITIATED" | "INITIALIZING" + ) +} + +/// Non-active rows safe to delete before starting a fresh Meta OAuth handoff. +pub fn is_clearable_oauth_status(status: &str) -> bool { + let upper = status.trim().to_ascii_uppercase(); + is_inflight_oauth_status(status) || matches!(upper.as_str(), "FAILED" | "ERROR" | "EXPIRED") +} + +/// Detect authorize-path failures that look like upstream rate limiting. +pub fn is_authorize_rate_limited(err: &str) -> bool { + let lower = err.to_ascii_lowercase(); + lower.contains("429") + || lower.contains("too many requests") + || lower.contains("rate limit") + || lower.contains("rate_limit") + || lower.contains("ratelimited") +} + +/// User-facing hint when Meta OAuth is rate-limited. +pub fn meta_oauth_rate_limit_message(toolkit: &str) -> String { + let name = toolkit.trim(); + let account_hint = if name.eq_ignore_ascii_case("instagram") { + " Use an Instagram Business or Creator account — personal accounts are not supported." + } else if name.eq_ignore_ascii_case("facebook") { + " Confirm the Facebook account has access to the relevant Page or Business Manager." + } else { + "" + }; + format!( + "Meta is temporarily rate-limiting {name} sign-in (HTTP 429). Wait a few \ + minutes before retrying and avoid clicking Connect repeatedly.{account_hint}" + ) +} + +/// If `err` is a Meta-toolkit authorize rate limit, replace it with guidance. +pub fn wrap_authorize_rate_limit_error(toolkit: &str, err: anyhow::Error) -> anyhow::Error { + let rendered = format!("{err:#}"); + if is_meta_oauth_toolkit(toolkit) && is_authorize_rate_limited(&rendered) { + anyhow::anyhow!("{}", meta_oauth_rate_limit_message(toolkit)) + } else { + err + } +} + +/// Remove non-active connection rows for `toolkit` so a fresh OAuth handoff does +/// not accumulate Meta sessions (#1952). +pub async fn clear_non_active_connections( + client: &ComposioClient, + toolkit: &str, +) -> anyhow::Result { + if !is_meta_oauth_toolkit(toolkit) { + return Ok(0); + } + let toolkit_key = toolkit.trim().to_ascii_lowercase(); + let resp = client.list_connections().await?; + let mut cleared = 0u32; + for conn in resp.connections { + if conn.normalized_toolkit() != toolkit_key { + continue; + } + if conn.is_active() || !is_clearable_oauth_status(&conn.status) { + continue; + } + tracing::info!( + toolkit = %toolkit_key, + connection_id = %conn.id, + status = %conn.status, + "[composio][oauth] clearing stale non-active connection before Meta OAuth handoff (#1952)" + ); + match client.delete_connection(&conn.id).await { + Ok(_) => cleared += 1, + Err(e) => { + tracing::warn!( + toolkit = %toolkit_key, + connection_id = %conn.id, + error = %e, + "[composio][oauth] failed to clear stale connection (non-fatal)" + ); + } + } + } + Ok(cleared) +} + +/// Begin a backend-proxied OAuth handoff with Meta cleanup + 429 backoff. +pub async fn authorize_with_meta_guard( + client: &ComposioClient, + toolkit: &str, + extra_params: Option, +) -> anyhow::Result { + let cleared = match clear_non_active_connections(client, toolkit).await { + Ok(cleared) => cleared, + Err(e) => { + tracing::warn!( + toolkit = %toolkit, + error = %e, + "[composio][oauth] pre-handoff cleanup failed; continuing authorize" + ); + 0 + } + }; + tracing::debug!( + toolkit = %toolkit, + cleared, + is_meta = is_meta_oauth_toolkit(toolkit), + "[composio][oauth] authorize_with_meta_guard: pre-handoff cleanup" + ); + authorize_with_rate_limit_retry(|| client.authorize(toolkit, extra_params.clone())).await +} + +/// Direct-mode authorize with the same 429 backoff used for Meta toolkits. +pub async fn direct_authorize_with_meta_guard( + direct: &std::sync::Arc, + toolkit: &str, + entity_id: &str, +) -> anyhow::Result { + authorize_with_rate_limit_retry(|| direct_authorize(direct, toolkit, entity_id)).await +} + +async fn authorize_with_rate_limit_retry( + mut attempt_authorize: F, +) -> anyhow::Result +where + F: FnMut() -> Fut, + Fut: std::future::Future>, +{ + let mut delay = AUTHORIZE_RATE_LIMIT_INITIAL_BACKOFF; + let mut last_err: Option = None; + for attempt in 1..=AUTHORIZE_RATE_LIMIT_MAX_ATTEMPTS { + match attempt_authorize().await { + Ok(resp) => return Ok(resp), + Err(e) => { + let rendered = format!("{e:#}"); + if is_authorize_rate_limited(&rendered) + && attempt < AUTHORIZE_RATE_LIMIT_MAX_ATTEMPTS + { + tracing::warn!( + attempt, + max_attempts = AUTHORIZE_RATE_LIMIT_MAX_ATTEMPTS, + sleep_secs = delay.as_secs(), + "[composio][oauth] authorize rate-limited; backing off (#1952)" + ); + tokio::time::sleep(delay).await; + delay = (delay * 2).min(AUTHORIZE_RATE_LIMIT_MAX_BACKOFF); + last_err = Some(e); + continue; + } + return Err(e); + } + } + } + Err(last_err.unwrap_or_else(|| anyhow::anyhow!("authorize failed after retries"))) +} + +#[cfg(test)] +#[path = "oauth_handoff_tests.rs"] +mod tests; diff --git a/src/openhuman/composio/oauth_handoff_tests.rs b/src/openhuman/composio/oauth_handoff_tests.rs new file mode 100644 index 000000000..9908eed86 --- /dev/null +++ b/src/openhuman/composio/oauth_handoff_tests.rs @@ -0,0 +1,128 @@ +//! Tests for Meta OAuth handoff helpers (#1952). + +use super::{ + authorize_with_meta_guard, is_authorize_rate_limited, is_clearable_oauth_status, + is_inflight_oauth_status, is_meta_oauth_toolkit, meta_oauth_rate_limit_message, + wrap_authorize_rate_limit_error, +}; +use axum::{ + http::StatusCode, + routing::{get, post}, + Json, Router, +}; +use serde_json::{json, Value}; +use std::sync::Arc; + +use super::ComposioClient; + +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(crate::openhuman::integrations::IntegrationClient::new( + base_url, + "test-token".into(), + )); + ComposioClient::new(inner) +} + +#[test] +fn meta_oauth_toolkit_detection() { + assert!(is_meta_oauth_toolkit("instagram")); + assert!(is_meta_oauth_toolkit("Facebook")); + assert!(!is_meta_oauth_toolkit("gmail")); +} + +#[test] +fn inflight_and_clearable_statuses() { + assert!(is_inflight_oauth_status("pending")); + assert!(is_inflight_oauth_status("INITIATED")); + assert!(!is_inflight_oauth_status("ACTIVE")); + + assert!(is_clearable_oauth_status("FAILED")); + assert!(is_clearable_oauth_status("EXPIRED")); + assert!(!is_clearable_oauth_status("ACTIVE")); +} + +#[test] +fn authorize_rate_limit_shape_detection() { + assert!(is_authorize_rate_limited( + "Backend returned 429 Too Many Requests" + )); + assert!(is_authorize_rate_limited("rate_limit exceeded")); + assert!(!is_authorize_rate_limited("401 Unauthorized")); +} + +#[test] +fn wrap_authorize_rate_limit_error_replaces_meta_toolkit_message() { + let err = anyhow::anyhow!("Backend returned 429 Too Many Requests"); + let wrapped = wrap_authorize_rate_limit_error("instagram", err); + let msg = format!("{wrapped:#}"); + assert!(msg.contains("Business or Creator")); + assert!(msg.contains("429")); +} + +#[test] +fn wrap_authorize_rate_limit_error_passthrough_for_non_meta() { + let err = anyhow::anyhow!("Backend returned 429 Too Many Requests"); + let wrapped = wrap_authorize_rate_limit_error("gmail", err); + assert!(format!("{wrapped:#}").contains("Backend returned 429")); +} + +#[test] +fn meta_oauth_rate_limit_message_mentions_business_account() { + let msg = meta_oauth_rate_limit_message("instagram"); + assert!(msg.to_ascii_lowercase().contains("business")); +} + +#[test] +fn meta_oauth_rate_limit_message_uses_facebook_specific_guidance() { + let msg = meta_oauth_rate_limit_message("facebook"); + assert!(msg.contains("Facebook")); + assert!(msg.contains("Business Manager")); + assert!(!msg.contains("Instagram Business or Creator")); +} + +#[tokio::test] +async fn authorize_continues_when_pre_handoff_cleanup_fails() { + let app = Router::new() + .route( + "/agent-integrations/composio/connections", + get(|| async { + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({ + "success": false, + "error": "temporary list failure" + })), + ) + }), + ) + .route( + "/agent-integrations/composio/authorize", + post(|Json(body): Json| async move { + assert_eq!(body["toolkit"].as_str(), Some("instagram")); + Json(json!({ + "success": true, + "data": { + "connectUrl": "https://composio.example/instagram/consent", + "connectionId": "conn-instagram" + } + })) + }), + ); + let client = build_client_for(start_mock_backend(app).await); + + let resp = authorize_with_meta_guard(&client, "instagram", None) + .await + .expect("authorize should continue when cleanup is unavailable"); + + assert_eq!(resp.connection_id, "conn-instagram"); + assert!(resp.connect_url.contains("instagram")); +} diff --git a/src/openhuman/composio/ops.rs b/src/openhuman/composio/ops.rs index e2f758633..0e2ef05ae 100644 --- a/src/openhuman/composio/ops.rs +++ b/src/openhuman/composio/ops.rs @@ -300,10 +300,13 @@ pub async fn composio_authorize( let resp = match kind { ComposioClientKind::Backend(client) => { tracing::debug!(toolkit = %toolkit, "[composio] authorize: backend variant"); - client.authorize(toolkit, extra_params).await.map_err(|e| { - report_composio_op_error("authorize", &e); - format!("[composio] authorize failed: {e:#}") - })? + super::oauth_handoff::authorize_with_meta_guard(&client, toolkit, extra_params) + .await + .map_err(|e| { + report_composio_op_error("authorize", &e); + let wrapped = super::oauth_handoff::wrap_authorize_rate_limit_error(toolkit, e); + format!("[composio] authorize failed: {wrapped:#}") + })? } ComposioClientKind::Direct(direct) => { tracing::info!( @@ -327,9 +330,16 @@ pub async fn composio_authorize( app.composio.dev for your auth config" ); } - direct_authorize(&direct, toolkit, &config.composio.entity_id) - .await - .map_err(|e| format!("[composio-direct] authorize failed: {e:#}"))? + super::oauth_handoff::direct_authorize_with_meta_guard( + &direct, + toolkit, + &config.composio.entity_id, + ) + .await + .map_err(|e| { + let wrapped = super::oauth_handoff::wrap_authorize_rate_limit_error(toolkit, e); + format!("[composio-direct] authorize failed: {wrapped:#}") + })? } }; diff --git a/src/openhuman/composio/ops_test.rs b/src/openhuman/composio/ops_test.rs index bc365380d..80fea1758 100644 --- a/src/openhuman/composio/ops_test.rs +++ b/src/openhuman/composio/ops_test.rs @@ -349,6 +349,60 @@ async fn composio_list_connections_via_mock_counts_active() { assert!(outcome.logs.iter().any(|l| l.contains("2 active"))); } +#[tokio::test] +async fn composio_authorize_clears_pending_meta_connection_before_handoff() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + let deletes = Arc::new(AtomicUsize::new(0)); + let deletes_for_delete = Arc::clone(&deletes); + let app = Router::new() + .route( + "/agent-integrations/composio/connections", + get(|| async { + Json(json!({ + "success": true, + "data": {"connections": [ + {"id":"ig-pending","toolkit":"instagram","status":"PENDING"} + ]} + })) + }), + ) + .route( + "/agent-integrations/composio/connections/{id}", + axum::routing::delete(move |Path(id): Path| { + let deletes = Arc::clone(&deletes_for_delete); + async move { + if id == "ig-pending" { + deletes.fetch_add(1, Ordering::SeqCst); + } + Json(json!({"success": true, "data": {"deleted": true}})) + } + }), + ) + .route( + "/agent-integrations/composio/authorize", + post(|Json(body): Json| async move { + assert_eq!(body["toolkit"], "instagram"); + Json(json!({ + "success": true, + "data": { + "connectUrl": "https://meta.example/oauth", + "connectionId": "c-new" + } + })) + }), + ); + let base = start_mock_backend(app).await; + let tmp = tempfile::tempdir().unwrap(); + let config = config_with_backend(&tmp, base); + let outcome = composio_authorize(&config, "instagram", None) + .await + .unwrap(); + assert_eq!(outcome.value.connection_id, "c-new"); + assert_eq!(deletes.load(Ordering::SeqCst), 1); +} + #[tokio::test] async fn composio_authorize_via_mock_publishes_event_and_returns_url() { let app = Router::new().route(