fix(integrations): distinguish mid-OAuth / expired / failed states in spawn gate (#2365) (#2373)

Co-authored-by: cyrus@tinyhumans.ai <cyrus@tinyhumans.ai>

src/openhuman/agent/agents/integrations_agent/prompt.rs

@@ -246,6 +246,7 @@ mod tests {
             tools: Vec::new(),
             gated_tools: Vec::new(),
             connected: true,
+            non_active_status: None,
         }];
         let body = build(&ctx_with(&integrations, &[])).unwrap();
         assert!(body.contains("## Connected Integrations"));
@@ -275,6 +276,7 @@ mod tests {
             tools: Vec::new(),
             gated_tools: Vec::new(),
             connected: false,
+            non_active_status: None,
         }];
         let body = build(&ctx_with(&integrations, &[])).unwrap();
         assert!(!body.contains("## Connected Integrations"));

src/openhuman/agent/agents/orchestrator/prompt.rs

@@ -237,6 +237,7 @@ mod tests {
             tools: Vec::new(),
             gated_tools: Vec::new(),
             connected: true,
+            non_active_status: None,
         }];
         let body = build(&ctx_with(&integrations)).unwrap();
         assert!(body.contains("## Connected Integrations"));
@@ -268,6 +269,7 @@ mod tests {
             tools: Vec::new(),
             gated_tools: Vec::new(),
             connected: true,
+            non_active_status: None,
         }];
         let body = build(&ctx_with(&integrations)).unwrap();
         assert!(body.contains("## Connected Integrations"));
@@ -290,6 +292,7 @@ mod tests {
                 tools: Vec::new(),
                 gated_tools: Vec::new(),
                 connected: true,
+                non_active_status: None,
             },
             ConnectedIntegration {
                 toolkit: "linear".into(),
@@ -297,6 +300,7 @@ mod tests {
                 tools: Vec::new(),
                 gated_tools: Vec::new(),
                 connected: false,
+                non_active_status: None,
             },
         ];
         let body = build(&ctx_with(&integrations)).unwrap();
@@ -312,6 +316,7 @@ mod tests {
             tools: Vec::new(),
             gated_tools: Vec::new(),
             connected: false,
+            non_active_status: None,
         }];
         let body = build(&ctx_with(&integrations)).unwrap();
         assert!(!body.contains("## Connected Integrations"));

src/openhuman/agent/agents/welcome/prompt.rs

@@ -181,6 +181,7 @@ mod tests {
                 tools: Vec::new(),
                 gated_tools: Vec::new(),
                 connected: true,
+                non_active_status: None,
             },
             ConnectedIntegration {
                 toolkit: "notion".into(),
@@ -188,6 +189,7 @@ mod tests {
                 tools: Vec::new(),
                 gated_tools: Vec::new(),
                 connected: false,
+                non_active_status: None,
             },
         ];
         let body = build(&ctx_with(&integrations)).unwrap();

src/openhuman/agent/harness/session/tests.rs

@@ -257,6 +257,7 @@ fn set_connected_integrations_marks_session_initialized_and_updates_hash() {
             tools: vec![],
             gated_tools: vec![],
             connected: true,
+            non_active_status: None,
         },
     ]);
 

src/openhuman/agent/harness/subagent_runner/ops.rs

@@ -727,6 +727,10 @@ async fn run_typed_mode(
                     // the user pref and doesn't change per-spawn.
                     gated_tools: cached_integration.gated_tools.clone(),
                     connected: cached_integration.connected,
+                    // Inherit the cached non-active status — this spawn
+                    // path only fires on connected toolkits, but keep the
+                    // field consistent with the source row for #2365.
+                    non_active_status: cached_integration.non_active_status.clone(),
                 };
                 let integration = &integration;
                 // Fuzzy-filter the toolkit's actions against the task prompt

src/openhuman/agent/harness/test_support_test.rs

@@ -1564,6 +1564,7 @@ async fn orchestrator_prompt_drives_composio_call_via_delegation_chain() {
         tools: Vec::new(),
         gated_tools: Vec::new(),
         connected: true,
+        non_active_status: None,
     }];
     let ctx = {
         use crate::openhuman::context::prompt::{LearnedContextData, ToolCallFormat};

src/openhuman/agent/prompts/types.rs

@@ -112,6 +112,19 @@ pub struct ConnectedIntegration {
     /// and the orchestrator must point the user at Settings instead of
     /// attempting to delegate.
     pub connected: bool,
+    /// Raw upstream connection status when a connection row exists but
+    /// is not `ACTIVE` — e.g. `"INITIATED"`, `"INITIALIZING"`,
+    /// `"FAILED"`, `"EXPIRED"`. `None` means either the user is
+    /// `ACTIVE` (use `connected = true`) OR there is no connection
+    /// row at all (truly disconnected).
+    ///
+    /// Used by the `integrations_agent` spawn-gate to surface the
+    /// real reason a delegation can't proceed — see issue #2365
+    /// ("Agent says Gmail is disconnected when sending email"). The
+    /// gate previously emitted the same "not authorized yet" message
+    /// regardless of whether OAuth was mid-flight, the token had
+    /// expired, or the user had simply never started the flow.
+    pub non_active_status: Option<String>,
 }
 
 /// A toolkit action that exists in the catalog but is currently hidden from

src/openhuman/composio/ops.rs

@@ -1678,6 +1678,86 @@ async fn fetch_connected_integrations_uncached(
         .filter(|toolkit| !toolkit.is_empty())
         .collect();
 
+    // Most-informative *non-active* status per toolkit slug. Lets the
+    // integrations_agent spawn-gate (#2365) emit a precise message
+    // when a connection row exists but isn't usable yet (`INITIATED`
+    // — OAuth still in progress) or any longer (`EXPIRED` / `FAILED`)
+    // — instead of the legacy generic "available but not authorized".
+    //
+    // Status priority (UI-actionability):
+    //   1. EXPIRED  — reconnect path
+    //   2. FAILED / ERROR — reconnect path
+    //   3. INITIATED / INITIALIZING / PENDING — finish OAuth in browser
+    //   4. anything else — passes through verbatim
+    let non_active_status_by_slug: std::collections::HashMap<String, String> = {
+        fn priority(status: &str) -> u8 {
+            let s = status.trim().to_ascii_uppercase();
+            match s.as_str() {
+                "EXPIRED" => 4,
+                "FAILED" | "ERROR" => 3,
+                "INITIATED" | "INITIALIZING" | "PENDING" => 2,
+                _ => 1,
+            }
+        }
+        let mut map: std::collections::HashMap<String, (u8, String)> =
+            std::collections::HashMap::new();
+        for conn in connections.iter().filter(|c| !c.is_active()) {
+            let slug = conn.normalized_toolkit();
+            if slug.is_empty() {
+                continue;
+            }
+            // Don't override an ACTIVE-slug — those carry no non-active
+            // status from this map's perspective.
+            if connected_slugs.contains(&slug) {
+                continue;
+            }
+            let p = priority(&conn.status);
+            map.entry(slug.clone())
+                .and_modify(|cur| {
+                    if p > cur.0 {
+                        tracing::debug!(
+                            target: "composio",
+                            toolkit = %slug,
+                            previous_status = %cur.1,
+                            previous_priority = cur.0,
+                            new_status = %conn.status,
+                            new_priority = p,
+                            "[composio] non_active_status_by_slug: upgraded most-informative status"
+                        );
+                        *cur = (p, conn.status.clone());
+                    } else {
+                        tracing::trace!(
+                            target: "composio",
+                            toolkit = %slug,
+                            kept_status = %cur.1,
+                            kept_priority = cur.0,
+                            candidate_status = %conn.status,
+                            candidate_priority = p,
+                            "[composio] non_active_status_by_slug: kept higher-priority status"
+                        );
+                    }
+                })
+                .or_insert_with(|| {
+                    tracing::debug!(
+                        target: "composio",
+                        toolkit = %slug,
+                        status = %conn.status,
+                        priority = p,
+                        "[composio] non_active_status_by_slug: first non-active row"
+                    );
+                    (p, conn.status.clone())
+                });
+        }
+        let final_map: std::collections::HashMap<String, String> =
+            map.into_iter().map(|(k, (_, v))| (k, v)).collect();
+        tracing::debug!(
+            target: "composio",
+            entries = final_map.len(),
+            "[composio] non_active_status_by_slug: final map"
+        );
+        final_map
+    };
+
     // Deduplicate the allowlist so a backend that returns duplicates
     // doesn't produce dual entries downstream.
     let mut unique_toolkits: Vec<String> = allowlisted_toolkits.clone();
@@ -1774,6 +1854,11 @@ async fn fetch_connected_integrations_uncached(
             tools,
             gated_tools,
             connected,
+            non_active_status: if connected {
+                None
+            } else {
+                non_active_status_by_slug.get(slug).cloned()
+            },
         });
     }
 
@@ -1789,6 +1874,7 @@ async fn fetch_connected_integrations_uncached(
         tracing::debug!(
             toolkit = %ci.toolkit,
             connected = ci.connected,
+            non_active_status = ?ci.non_active_status,
             tool_count = ci.tools.len(),
             "[composio] integration overview"
         );

src/openhuman/composio/ops_test.rs

@@ -926,6 +926,7 @@ fn integration(toolkit: &str, connected: bool) -> ConnectedIntegration {
         tools: Vec::new(),
         gated_tools: Vec::new(),
         connected,
+        non_active_status: None,
     }
 }
 

src/openhuman/tools/impl/agent/spawn_subagent.rs

@@ -352,13 +352,27 @@ impl Tool for SpawnSubagentTool {
                             //      doesn't render this as a failed tool call.
                             // The model still reads the explanation and produces
                             // an appropriate user-facing response.
-                            return Ok(ToolResult::success(format!(
-                                "Integration '{tk}' is available but the user has not \
-                                 authorized it yet. Do NOT retry this spawn. Tell the user \
-                                 the integration is available and ask them to authorize \
-                                 '{tk}' in Connections → Integrations before retrying the \
-                                 original request."
-                            )));
+                            //
+                            // Split (#2365) into 4 cases driven by the upstream
+                            // status field on the most-informative connection
+                            // row, instead of the legacy generic
+                            // "not authorized yet" copy. Before this split,
+                            // an OAuth-in-progress / expired / failed Gmail
+                            // surfaced the same "you need to connect Gmail"
+                            // message — which Settings UI contradicted (it
+                            // shows the connection as initiated/expired), so
+                            // users concluded the agent was confused.
+                            tracing::debug!(
+                                target: "spawn_subagent",
+                                toolkit = %ci.toolkit,
+                                non_active_status = ?ci.non_active_status,
+                                "[spawn_subagent] integrations_agent gate: toolkit not connected — emitting status-specific message"
+                            );
+                            let message = describe_unconnected_state(
+                                &ci.toolkit,
+                                ci.non_active_status.as_deref(),
+                            );
+                            return Ok(ToolResult::success(message));
                         }
                         Some(_) => {
                             tracing::debug!(
@@ -631,6 +645,84 @@ fn render_worker_thread_result(
     )
 }
 
+/// Build the user-facing explanation for an allowlisted-but-not-active
+/// integration during an `integrations_agent` spawn (#2365).
+///
+/// The single message that previously covered every cause ("available
+/// but the user has not authorized it yet") looked confused to users
+/// who had Gmail showing in Settings (because Settings reflects the
+/// FE's optimistic post-OAuth view, while the spawn gate reads the
+/// backend's authoritative status). We now pivot on the upstream
+/// connection status:
+///
+/// - `INITIATED` / `INITIALIZING` / `PENDING` — OAuth in progress;
+///   ask the user to finish the flow in their browser.
+/// - `EXPIRED` — token rolled over; reconnect.
+/// - `FAILED` / `ERROR` — handshake didn't land; reconnect.
+/// - any other non-active status — quote the upstream verbatim.
+/// - `None` — no connection row at all (truly disconnected).
+///
+/// Returns text the model reads literally; the orchestrator paraphrases
+/// it into a user-facing reply. Keep the *intent* stable across
+/// rewordings — the "Settings → Connections → {toolkit}" path is
+/// load-bearing for the UI navigation tests.
+pub(crate) fn describe_unconnected_state(toolkit: &str, status: Option<&str>) -> String {
+    // Keep the original (trimmed) status separately so the
+    // unknown-status branch can quote it verbatim — CodeRabbit
+    // review on #2373: matching on the uppercased value AND
+    // formatting with that uppercased value broke the
+    // "quote upstream status verbatim" contract for mixed/lowercase
+    // wire shapes.
+    let trimmed = status.map(str::trim).filter(|s| !s.is_empty());
+    let upper = trimmed.map(|s| s.to_ascii_uppercase());
+    match upper.as_deref() {
+        Some("INITIATED") | Some("INITIALIZING") | Some("PENDING") => format!(
+            "Integration '{toolkit}' has an OAuth flow in progress but it hasn't reached \
+             ACTIVE yet. Do NOT retry this spawn. Tell the user the authorization is \
+             pending and ask them to finish the browser OAuth flow (Settings → \
+             Connections → '{toolkit}') before retrying. If they already closed the \
+             browser tab, they can restart the connection from the same Settings page."
+        ),
+        Some("EXPIRED") => format!(
+            "Integration '{toolkit}' is connected but the OAuth token has expired. \
+             Do NOT retry this spawn. Tell the user the connection expired and ask \
+             them to reconnect '{toolkit}' at Settings → Connections → '{toolkit}' \
+             before retrying the original request."
+        ),
+        Some("FAILED") | Some("ERROR") => {
+            // Quote the actual upstream label (FAILED / ERROR) instead of
+            // hard-coding "FAILED" — triage cross-references backend logs
+            // and a misquoted `ERROR` row showing up as "FAILED" wastes
+            // their time. graycyrus review on #2373.
+            let raw = trimmed.unwrap_or("");
+            format!(
+                "Integration '{toolkit}' has a previous OAuth attempt in a `{raw}` state. \
+                 Do NOT retry this spawn. Tell the user the connection failed and ask them \
+                 to reconnect '{toolkit}' at Settings → Connections → '{toolkit}' before \
+                 retrying the original request."
+            )
+        }
+        Some(_) => {
+            // Quote the *original* upstream status, not its uppercased
+            // form — preserves "DeauthRequired" / "needs_relink"-style
+            // mixed-case wire values for triage.
+            let raw = trimmed.unwrap_or("");
+            format!(
+                "Integration '{toolkit}' has a connection row but its status is `{raw}`, \
+                 which is not yet usable. Do NOT retry this spawn. Tell the user the \
+                 connection is in an unusable state and ask them to reconnect '{toolkit}' \
+                 at Settings → Connections → '{toolkit}'."
+            )
+        }
+        _ => format!(
+            "Integration '{toolkit}' is available but the user has not authorized it \
+             yet. Do NOT retry this spawn. Tell the user the integration is available \
+             and ask them to authorize '{toolkit}' in Settings → Connections → \
+             '{toolkit}' before retrying the original request."
+        ),
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -887,4 +979,130 @@ mod tests {
         )));
         assert!(out.contains("Valid toolkits"));
     }
+
+    // ── #2365: describe_unconnected_state per upstream status ───────
+
+    #[test]
+    fn describe_unconnected_state_initiated_says_oauth_in_progress() {
+        let msg = describe_unconnected_state("gmail", Some("INITIATED"));
+        assert!(
+            msg.contains("OAuth flow in progress"),
+            "INITIATED must surface the in-progress wording: {msg}"
+        );
+        assert!(msg.contains("Settings → Connections → 'gmail'"));
+        // The legacy "not authorized yet" copy must NOT leak into the
+        // pending-OAuth branch — that was the user-perception bug
+        // from #2365 (Settings UI showed Gmail connected, agent said
+        // "not authorized").
+        assert!(
+            !msg.contains("has not authorized it yet"),
+            "INITIATED must not borrow the truly-disconnected copy: {msg}"
+        );
+    }
+
+    #[test]
+    fn describe_unconnected_state_pending_and_initializing_are_aliased() {
+        for status in ["PENDING", "INITIALIZING"] {
+            let msg = describe_unconnected_state("gmail", Some(status));
+            assert!(
+                msg.contains("OAuth flow in progress"),
+                "{status} must hit the in-progress branch: {msg}"
+            );
+        }
+    }
+
+    #[test]
+    fn describe_unconnected_state_expired_says_reconnect() {
+        let msg = describe_unconnected_state("gmail", Some("EXPIRED"));
+        assert!(msg.contains("OAuth token has expired"));
+        assert!(msg.contains("reconnect 'gmail'"));
+        assert!(!msg.contains("OAuth flow in progress"));
+    }
+
+    #[test]
+    fn describe_unconnected_state_failed_and_error_route_to_reconnect() {
+        for status in ["FAILED", "ERROR"] {
+            let msg = describe_unconnected_state("gmail", Some(status));
+            let expected = format!("`{status}` state");
+            assert!(
+                msg.contains(&expected),
+                "{status} must be quoted verbatim, not collapsed to a single label: {msg}"
+            );
+            assert!(msg.contains("reconnect 'gmail'"));
+        }
+    }
+
+    #[test]
+    fn describe_unconnected_state_failed_and_error_preserve_original_casing() {
+        // Mixed-case wire values must round-trip through the FAILED /
+        // ERROR branch with their original casing intact — that's the
+        // whole point of graycyrus' review feedback.
+        let lower_failed = describe_unconnected_state("gmail", Some("failed"));
+        assert!(
+            lower_failed.contains("`failed` state"),
+            "lowercase `failed` must be quoted verbatim: {lower_failed}"
+        );
+        let mixed_error = describe_unconnected_state("gmail", Some("Error"));
+        assert!(
+            mixed_error.contains("`Error` state"),
+            "mixed-case `Error` must be quoted verbatim: {mixed_error}"
+        );
+    }
+
+    #[test]
+    fn describe_unconnected_state_quotes_unknown_status_verbatim() {
+        // Pin three shapes (uppercase / mixed / snake_case) so the
+        // verbatim-quoting contract can't silently drift back to
+        // echoing the matched (uppercased) value — that was the
+        // CodeRabbit finding on #2373.
+        for raw in ["DEAUTH_REQUIRED", "needs_relink", "PartialAuthRequired"] {
+            let msg = describe_unconnected_state("gmail", Some(raw));
+            let expected = format!("`{raw}`");
+            assert!(
+                msg.contains(&expected),
+                "unknown status `{raw}` must be quoted verbatim (not its uppercased form): {msg}"
+            );
+        }
+    }
+
+    #[test]
+    fn describe_unconnected_state_quotes_unknown_status_after_trimming_whitespace() {
+        // Whitespace-only / blank statuses must NOT hit the
+        // unknown-status branch — they collapse to the
+        // truly-disconnected legacy copy via the `filter(|s|
+        // !s.is_empty())` guard in `describe_unconnected_state`.
+        let blank = describe_unconnected_state("gmail", Some("   "));
+        assert!(
+            blank.contains("has not authorized it yet"),
+            "whitespace-only status must collapse to legacy None branch: {blank}"
+        );
+        // A real status with surrounding whitespace is quoted with
+        // the whitespace trimmed (not preserved verbatim — triage
+        // would not want padded backticks).
+        let padded = describe_unconnected_state("gmail", Some("  DeauthRequired  "));
+        assert!(
+            padded.contains("`DeauthRequired`"),
+            "trimmed status must be quoted in original casing: {padded}"
+        );
+    }
+
+    #[test]
+    fn describe_unconnected_state_none_is_truly_disconnected() {
+        let msg = describe_unconnected_state("gmail", None);
+        assert!(
+            msg.contains("has not authorized it yet"),
+            "None must hit the legacy never-connected copy: {msg}"
+        );
+        assert!(msg.contains("Settings → Connections → 'gmail'"));
+    }
+
+    #[test]
+    fn describe_unconnected_state_status_match_is_case_insensitive() {
+        // The status string flows in from Composio's wire format; we
+        // can't assume casing. The classifier must normalise.
+        let initiated = describe_unconnected_state("gmail", Some("initiated"));
+        assert!(initiated.contains("OAuth flow in progress"));
+        let expired = describe_unconnected_state("gmail", Some("Expired"));
+        assert!(expired.contains("OAuth token has expired"));
+    }
 }

src/openhuman/tools/orchestrator_tools.rs

@@ -322,6 +322,7 @@ mod tests {
             tools: vec![],
             gated_tools: vec![],
             connected: true,
+            non_active_status: None,
         }
     }
 
@@ -503,6 +504,7 @@ mod tests {
                 tools: vec![],
                 gated_tools: vec![],
                 connected: false, // not connected — must not appear in the enum
+                non_active_status: None,
             },
             integration("notion", "Read and write pages."),
         ];
@@ -577,6 +579,7 @@ mod tests {
                 tools: vec![],
                 gated_tools: vec![],
                 connected: true,
+                non_active_status: None,
             },
             integration("gmail", "Email."),
         ];