fix(composio): stop direct-mode invalid key polling (#4318)

app/scripts/e2e-run-session.sh

@@ -174,6 +174,8 @@ export CEF_CDP_PORT
 # rest of the mock backend.  The core reads this at TelegramChannel::new() time,
 # which runs after the config is fully loaded.
 export OPENHUMAN_TELEGRAM_BOT_API_BASE="http://127.0.0.1:${E2E_MOCK_PORT}"
+export OPENHUMAN_COMPOSIO_DIRECT_BASE_V2="http://127.0.0.1:${E2E_MOCK_PORT}"
+export OPENHUMAN_COMPOSIO_DIRECT_BASE_V3="http://127.0.0.1:${E2E_MOCK_PORT}"
 
 echo "[runner] Killing any running OpenHuman instances..."
 case "$OS" in

app/scripts/e2e-web-session.sh

@@ -129,6 +129,8 @@ fi
 
 export OPENHUMAN_CORE_TOKEN="$PW_CORE_RPC_TOKEN"
 export OPENHUMAN_TELEGRAM_BOT_API_BASE="http://127.0.0.1:${E2E_MOCK_PORT}"
+export OPENHUMAN_COMPOSIO_DIRECT_BASE_V2="http://127.0.0.1:${E2E_MOCK_PORT}"
+export OPENHUMAN_COMPOSIO_DIRECT_BASE_V3="http://127.0.0.1:${E2E_MOCK_PORT}"
 # Keep the standalone core aligned with the Rust mock runner: sub-agent
 # orchestration builds large async futures and can overflow the default stack.
 export RUST_MIN_STACK="${RUST_MIN_STACK:-16777216}"

scripts/mock-api/routes/integrations.mjs

@@ -197,6 +197,15 @@ export function handleIntegrations(ctx) {
   // (chat/completions is handled by routes/llm.mjs ahead of this route)
 
   // ── Composio ───────────────────────────────────────────────
+  if (
+    method === "GET" &&
+    /^\/(?:api\/v3\/)?connected_accounts\/?(\?.*)?$/.test(url)
+  ) {
+    const items = parseBehaviorJson("composioDirectConnectedAccounts", []);
+    json(res, 200, { items });
+    return true;
+  }
+
   if (
     method === "GET" &&
     /^\/agent-integrations\/composio\/toolkits\/?(\?.*)?$/.test(url)

src/openhuman/composio/client.rs

@@ -705,6 +705,54 @@ pub enum ComposioClientKind {
     Direct(Arc<crate::openhuman::tools::ComposioTool>),
 }
 
+pub(crate) fn create_direct_composio_tool_for_api_key(
+    config: &crate::openhuman::config::Config,
+    api_key: &str,
+) -> anyhow::Result<Arc<crate::openhuman::tools::ComposioTool>> {
+    let api_key = api_key.trim();
+    if api_key.is_empty() {
+        anyhow::bail!("composio direct api key must not be empty");
+    }
+
+    // The direct client takes a `SecurityPolicy` for `Tool::execute`
+    // gating, but the factory's job is only to materialize a *client*
+    // — it does not actually invoke `execute()` itself, so the
+    // default policy is sufficient here. Callers that go through
+    // the `Tool` surface re-acquire the live policy from their own
+    // context.
+    let security = Arc::new(crate::openhuman::security::SecurityPolicy::default());
+    #[cfg(debug_assertions)]
+    let tool = match (
+        std::env::var("OPENHUMAN_COMPOSIO_DIRECT_BASE_V2").ok(),
+        std::env::var("OPENHUMAN_COMPOSIO_DIRECT_BASE_V3").ok(),
+    ) {
+        (Some(base_v2), Some(base_v3)) => {
+            crate::openhuman::tools::ComposioTool::new_with_base_urls_for_loopback(
+                api_key,
+                Some(config.composio.entity_id.as_str()),
+                security,
+                base_v2,
+                base_v3,
+            )
+            .map_err(|e| {
+                anyhow::anyhow!("invalid debug composio direct loopback base override: {e}")
+            })?
+        }
+        _ => crate::openhuman::tools::ComposioTool::new(
+            api_key,
+            Some(config.composio.entity_id.as_str()),
+            security,
+        ),
+    };
+    #[cfg(not(debug_assertions))]
+    let tool = crate::openhuman::tools::ComposioTool::new(
+        api_key,
+        Some(config.composio.entity_id.as_str()),
+        security,
+    );
+    Ok(Arc::new(tool))
+}
+
 impl ComposioClientKind {
     /// Returns `"backend"` or `"direct"` — handy for logging and tests.
     pub fn mode(&self) -> &'static str {
@@ -772,47 +820,12 @@ pub fn create_composio_client(
                     )
                 })?;
 
-            // The direct client takes a `SecurityPolicy` for `Tool::execute`
-            // gating, but the factory's job is only to materialize a *client*
-            // — it does not actually invoke `execute()` itself, so the
-            // default policy is sufficient here. Callers that go through
-            // the `Tool` surface re-acquire the live policy from their own
-            // context.
-            let security = Arc::new(crate::openhuman::security::SecurityPolicy::default());
-            #[cfg(debug_assertions)]
-            let tool = match (
-                std::env::var("OPENHUMAN_COMPOSIO_DIRECT_BASE_V2").ok(),
-                std::env::var("OPENHUMAN_COMPOSIO_DIRECT_BASE_V3").ok(),
-            ) {
-                (Some(base_v2), Some(base_v3)) => {
-                    crate::openhuman::tools::ComposioTool::new_with_base_urls_for_loopback(
-                        &api_key,
-                        Some(config.composio.entity_id.as_str()),
-                        security,
-                        base_v2,
-                        base_v3,
-                    )
-                    .map_err(|e| {
-                        anyhow::anyhow!("invalid debug composio direct loopback base override: {e}")
-                    })?
-                }
-                _ => crate::openhuman::tools::ComposioTool::new(
-                    &api_key,
-                    Some(config.composio.entity_id.as_str()),
-                    security,
-                ),
-            };
-            #[cfg(not(debug_assertions))]
-            let tool = crate::openhuman::tools::ComposioTool::new(
-                &api_key,
-                Some(config.composio.entity_id.as_str()),
-                security,
-            );
+            let tool = create_direct_composio_tool_for_api_key(config, &api_key)?;
             tracing::debug!(
                 key_len = api_key.len(),
                 "[composio-factory] resolved direct variant (key redacted)"
             );
-            Ok(ComposioClientKind::Direct(Arc::new(tool)))
+            Ok(ComposioClientKind::Direct(tool))
         }
         unknown => {
             tracing::warn!(mode = %unknown, "[composio-factory] unknown composio mode");
@@ -836,7 +849,7 @@ pub fn create_composio_client(
 // direct-mode plumbing can see the full envelope-translation surface
 // in one place.
 
-use super::types::ComposioConnection;
+use super::{direct_auth, types::ComposioConnection};
 
 /// Direct-mode counterpart to [`ComposioClient::authorize`]. Calls
 /// Composio v3 `/connected_accounts/link` via
@@ -963,7 +976,44 @@ pub async fn direct_list_connections(
     direct: &Arc<crate::openhuman::tools::ComposioTool>,
 ) -> anyhow::Result<ComposioConnectionsResponse> {
     tracing::debug!("[composio-direct] list_connections: GET v3 /connected_accounts");
-    let items = direct.list_connected_accounts().await?;
+    let key_id = direct.auth_key_fingerprint();
+    if let Some(error) = direct_auth::direct_auth_backoff_error(key_id) {
+        tracing::warn!(
+            "[composio-direct] list_connections: direct API key backoff gate open; \
+             skipping v3 /connected_accounts"
+        );
+        anyhow::bail!("{error}");
+    }
+
+    let items = match direct.list_connected_accounts().await {
+        Ok(items) => {
+            direct_auth::record_direct_auth_success(key_id);
+            items
+        }
+        Err(error) => {
+            let rendered = format!("{error:#}");
+            match direct_auth::record_direct_auth_failure(key_id, &rendered) {
+                direct_auth::DirectAuthFailureDecision::NotAuthFailure => {}
+                direct_auth::DirectAuthFailureDecision::RetryAllowed { consecutive } => {
+                    tracing::warn!(
+                        consecutive,
+                        threshold = direct_auth::DIRECT_INVALID_API_KEY_THRESHOLD,
+                        "[composio-direct] list_connections: direct API key rejected"
+                    );
+                }
+                direct_auth::DirectAuthFailureDecision::CircuitOpened { consecutive } => {
+                    let backoff = direct_auth::invalid_api_key_backoff_message(consecutive);
+                    tracing::warn!(
+                        consecutive,
+                        threshold = direct_auth::DIRECT_INVALID_API_KEY_THRESHOLD,
+                        "[composio-direct] list_connections: direct API key backoff gate opened"
+                    );
+                    anyhow::bail!("{backoff}");
+                }
+            }
+            return Err(error);
+        }
+    };
     let connections: Vec<ComposioConnection> = items
         .into_iter()
         .filter_map(|item| {

src/openhuman/composio/client_tests.rs

@@ -1142,14 +1142,90 @@ fn direct_tool_for_test() -> std::sync::Arc<crate::openhuman::tools::ComposioToo
 /// Like [`direct_tool_for_test`] but with the v3 base pointed at a local
 /// mock server so HTTP paths (e.g. `direct_list_tools`) can be asserted.
 fn direct_tool_for_mock(base_v3: String) -> std::sync::Arc<crate::openhuman::tools::ComposioTool> {
+    direct_tool_for_mock_with_key(base_v3, "ck_test_direct")
+}
+
+fn direct_tool_for_mock_with_key(
+    base_v3: String,
+    api_key: &str,
+) -> std::sync::Arc<crate::openhuman::tools::ComposioTool> {
     std::sync::Arc::new(crate::openhuman::tools::ComposioTool::new_with_v3_base(
-        "ck_test_direct",
+        api_key,
         Some("default"),
         std::sync::Arc::new(crate::openhuman::security::SecurityPolicy::default()),
         base_v3,
     ))
 }
 
+struct DirectAuthFailureGuard {
+    key_id: u64,
+}
+
+impl DirectAuthFailureGuard {
+    fn for_tool(tool: &std::sync::Arc<crate::openhuman::tools::ComposioTool>) -> Self {
+        let key_id = tool.auth_key_fingerprint();
+        crate::openhuman::composio::direct_auth::reset_direct_auth_failure(key_id);
+        Self { key_id }
+    }
+}
+
+impl Drop for DirectAuthFailureGuard {
+    fn drop(&mut self) {
+        crate::openhuman::composio::direct_auth::reset_direct_auth_failure(self.key_id);
+    }
+}
+
+#[tokio::test]
+async fn direct_list_connections_stops_hitting_composio_after_repeated_invalid_api_key() {
+    let hits = Arc::new(AtomicUsize::new(0));
+    let app = Router::new()
+        .route(
+            "/connected_accounts",
+            get(|State(hits): State<Arc<AtomicUsize>>| async move {
+                hits.fetch_add(1, Ordering::SeqCst);
+                (
+                    StatusCode::UNAUTHORIZED,
+                    Json(json!({ "error": { "message": "Invalid API key" } })),
+                )
+            }),
+        )
+        .with_state(hits.clone());
+    let base = start_mock_backend(app).await;
+    let tool = direct_tool_for_mock_with_key(base, "ck_test_direct_invalid_backoff");
+    let _auth_guard = DirectAuthFailureGuard::for_tool(&tool);
+
+    for _ in 0..2 {
+        let err = direct_list_connections(&tool)
+            .await
+            .expect_err("invalid key should reject");
+        assert!(
+            err.to_string().contains("Invalid API key"),
+            "unexpected error: {err:#}"
+        );
+    }
+
+    let opened = direct_list_connections(&tool)
+        .await
+        .expect_err("third invalid-key failure should open the backoff gate");
+    assert!(
+        opened.to_string().contains("re-enter"),
+        "backoff error should be actionable, got: {opened:#}"
+    );
+
+    let short_circuit = direct_list_connections(&tool)
+        .await
+        .expect_err("open backoff gate should short-circuit before HTTP");
+    assert!(
+        short_circuit.to_string().contains("re-enter"),
+        "short-circuit error should stay actionable, got: {short_circuit:#}"
+    );
+    assert_eq!(
+        hits.load(Ordering::SeqCst),
+        3,
+        "after three invalid-key failures, later polls must not hit Composio"
+    );
+}
+
 #[tokio::test]
 async fn direct_authorize_rejects_empty_toolkit() {
     let tool = direct_tool_for_test();

src/openhuman/composio/direct_auth/mod.rs

@@ -0,0 +1,125 @@
+//! Direct-mode Composio API-key health tracking.
+//!
+//! Invalid/revoked BYO keys used to make every 5s poll hit Composio v3 and
+//! fail again. This module keeps a process-local consecutive-failure counter
+//! keyed by a non-logged key fingerprint so repeated `401 Invalid API key`
+//! responses open a short-circuit gate until the user re-enters the key.
+
+use std::collections::HashMap;
+use std::hash::{Hash, Hasher};
+use std::sync::{LazyLock, Mutex, MutexGuard};
+
+pub(crate) const DIRECT_INVALID_API_KEY_THRESHOLD: u32 = 3;
+
+static DIRECT_AUTH_FAILURES: LazyLock<Mutex<HashMap<u64, u32>>> =
+    LazyLock::new(|| Mutex::new(HashMap::new()));
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum DirectAuthFailureDecision {
+    NotAuthFailure,
+    RetryAllowed { consecutive: u32 },
+    CircuitOpened { consecutive: u32 },
+}
+
+fn failure_counts() -> MutexGuard<'static, HashMap<u64, u32>> {
+    DIRECT_AUTH_FAILURES
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner())
+}
+
+pub(crate) fn fingerprint_api_key(api_key: &str) -> u64 {
+    let mut hasher = std::collections::hash_map::DefaultHasher::new();
+    api_key.trim().hash(&mut hasher);
+    hasher.finish()
+}
+
+pub(crate) fn is_invalid_api_key_error(message: &str) -> bool {
+    let lower = message.to_ascii_lowercase();
+    lower.contains("invalid api key")
+        || (lower.contains("401") && lower.contains("api key") && lower.contains("invalid"))
+}
+
+pub(crate) fn record_direct_auth_success(key_id: u64) {
+    failure_counts().remove(&key_id);
+}
+
+pub(crate) fn reset_direct_auth_failure(key_id: u64) {
+    failure_counts().remove(&key_id);
+}
+
+pub(crate) fn reset_all_direct_auth_failures() {
+    failure_counts().clear();
+}
+
+pub(crate) fn record_direct_auth_failure(key_id: u64, message: &str) -> DirectAuthFailureDecision {
+    if !is_invalid_api_key_error(message) {
+        reset_direct_auth_failure(key_id);
+        return DirectAuthFailureDecision::NotAuthFailure;
+    }
+
+    let mut counts = failure_counts();
+    let consecutive = counts.entry(key_id).or_insert(0);
+    *consecutive = consecutive.saturating_add(1);
+    if *consecutive >= DIRECT_INVALID_API_KEY_THRESHOLD {
+        DirectAuthFailureDecision::CircuitOpened {
+            consecutive: *consecutive,
+        }
+    } else {
+        DirectAuthFailureDecision::RetryAllowed {
+            consecutive: *consecutive,
+        }
+    }
+}
+
+pub(crate) fn direct_auth_backoff_error(key_id: u64) -> Option<String> {
+    let counts = failure_counts();
+    let consecutive = counts.get(&key_id).copied().unwrap_or_default();
+    (consecutive >= DIRECT_INVALID_API_KEY_THRESHOLD)
+        .then(|| invalid_api_key_backoff_message(consecutive))
+}
+
+pub(crate) fn invalid_api_key_backoff_message(consecutive: u32) -> String {
+    format!(
+        "Direct-mode Composio API key was rejected {consecutive} consecutive times with HTTP 401 Invalid API key; re-enter a valid key in Settings > Connections > Composio to resume polling."
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn generic_http_401_is_not_classified_as_invalid_api_key() {
+        assert!(!is_invalid_api_key_error("HTTP 401"));
+        assert!(is_invalid_api_key_error("HTTP 401: Invalid API key"));
+    }
+
+    #[test]
+    fn non_auth_failure_resets_invalid_key_streak() {
+        let key_id = fingerprint_api_key("ck_test_streak_reset");
+        reset_direct_auth_failure(key_id);
+
+        assert_eq!(
+            record_direct_auth_failure(key_id, "HTTP 401: Invalid API key"),
+            DirectAuthFailureDecision::RetryAllowed { consecutive: 1 }
+        );
+        assert_eq!(
+            record_direct_auth_failure(key_id, "HTTP 401: Invalid API key"),
+            DirectAuthFailureDecision::RetryAllowed { consecutive: 2 }
+        );
+        assert_eq!(
+            record_direct_auth_failure(key_id, "HTTP 500: upstream unavailable"),
+            DirectAuthFailureDecision::NotAuthFailure
+        );
+        assert!(
+            direct_auth_backoff_error(key_id).is_none(),
+            "non-auth failures must clear stale invalid-key counts"
+        );
+        assert_eq!(
+            record_direct_auth_failure(key_id, "HTTP 401: Invalid API key"),
+            DirectAuthFailureDecision::RetryAllowed { consecutive: 1 }
+        );
+
+        reset_direct_auth_failure(key_id);
+    }
+}

src/openhuman/composio/mod.rs

@@ -40,6 +40,7 @@ pub mod auth_retry;
 pub mod bus;
 pub mod client;
 mod connected_integrations;
+pub(crate) mod direct_auth;
 pub mod error_mapping;
 pub mod execute_dispatch;
 pub mod execute_prepare;

src/openhuman/composio/ops/direct_mode.rs

@@ -3,8 +3,44 @@
 use crate::openhuman::config::Config;
 use crate::rpc::RpcOutcome;
 
+use super::super::{
+    client::{create_direct_composio_tool_for_api_key, direct_list_connections},
+    direct_auth,
+};
 use super::error_utils::OpResult;
 
+async fn validate_direct_api_key_before_store(config: &Config, api_key: &str) -> OpResult<()> {
+    let key_id = direct_auth::fingerprint_api_key(api_key);
+    direct_auth::reset_direct_auth_failure(key_id);
+
+    let direct = create_direct_composio_tool_for_api_key(config, api_key)
+        .map_err(|e| format!("[composio-direct] validate_api_key: {e}"))?;
+
+    match direct_list_connections(&direct).await {
+        Ok(_) => {
+            direct_auth::record_direct_auth_success(key_id);
+            tracing::debug!("[composio-direct] validate_api_key: probe succeeded");
+            Ok(())
+        }
+        Err(error) => {
+            let rendered = format!("{error:#}");
+            if direct_auth::is_invalid_api_key_error(&rendered) {
+                tracing::warn!(
+                    "[composio-direct] validate_api_key: rejected invalid API key before storage"
+                );
+                Err("Invalid Composio API key. Re-enter a valid key in Settings > Connections > Composio.".into())
+            } else {
+                tracing::warn!(
+                    error = %rendered,
+                    "[composio-direct] validate_api_key: probe failed for a non-auth reason; \
+                     storing key so the user can retry when Composio is reachable"
+                );
+                Ok(())
+            }
+        }
+    }
+}
+
 /// Read the current Composio routing mode and whether a direct-mode API
 /// key is stored. **The key itself is never returned** — only a boolean
 /// flag so the UI can show a "Connected" / "Not set" status.
@@ -49,6 +85,7 @@ pub async fn composio_set_api_key(
         activate_direct,
         "[composio-direct] set_api_key (redacted)"
     );
+    validate_direct_api_key_before_store(config, trimmed).await?;
 
     crate::openhuman::credentials::store_composio_api_key(config, trimmed)
         .await
@@ -110,6 +147,7 @@ pub async fn composio_clear_api_key(config: &Config) -> OpResult<RpcOutcome<serd
         .save()
         .await
         .map_err(|e| format!("[composio-direct] save config failed: {e}"))?;
+    direct_auth::reset_all_direct_auth_failures();
 
     crate::core::event_bus::publish_global(
         crate::core::event_bus::DomainEvent::ComposioConfigChanged {

src/openhuman/composio/ops_tests.rs

@@ -236,7 +236,8 @@ use crate::openhuman::memory_store::chunks::types::{
     chunk_id, Chunk, Metadata, SourceKind, SourceRef,
 };
 use axum::{
-    extract::{Path, Query},
+    extract::{Path, Query, State},
+    http::HeaderMap,
     routing::{get, post},
     Json, Router,
 };
@@ -271,6 +272,52 @@ impl Drop for WorkspaceEnvGuard {
     }
 }
 
+struct EnvVarGuard {
+    key: &'static str,
+    previous: Option<std::ffi::OsString>,
+}
+
+impl EnvVarGuard {
+    fn set(key: &'static str, value: &str) -> Self {
+        let previous = std::env::var_os(key);
+        unsafe {
+            std::env::set_var(key, value);
+        }
+        Self { key, previous }
+    }
+}
+
+impl Drop for EnvVarGuard {
+    fn drop(&mut self) {
+        match self.previous.take() {
+            Some(prev) => unsafe {
+                std::env::set_var(self.key, prev);
+            },
+            None => unsafe {
+                std::env::remove_var(self.key);
+            },
+        }
+    }
+}
+
+struct DirectAuthFailureGuard {
+    key_id: u64,
+}
+
+impl DirectAuthFailureGuard {
+    fn new(api_key: &str) -> Self {
+        let key_id = crate::openhuman::composio::direct_auth::fingerprint_api_key(api_key);
+        crate::openhuman::composio::direct_auth::reset_direct_auth_failure(key_id);
+        Self { key_id }
+    }
+}
+
+impl Drop for DirectAuthFailureGuard {
+    fn drop(&mut self) {
+        crate::openhuman::composio::direct_auth::reset_direct_auth_failure(self.key_id);
+    }
+}
+
 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();
@@ -1926,6 +1973,104 @@ async fn composio_list_connections_returns_empty_when_direct_mode_no_key() {
     );
 }
 
+#[tokio::test]
+async fn composio_set_api_key_rejects_invalid_direct_key_before_persisting() {
+    use crate::openhuman::config::TEST_ENV_LOCK;
+    use crate::openhuman::credentials::get_composio_api_key;
+
+    let _env_guard = TEST_ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+    let app = Router::new().route(
+        "/connected_accounts",
+        get(|| async {
+            (
+                axum::http::StatusCode::UNAUTHORIZED,
+                Json(json!({ "error": { "message": "Invalid API key" } })),
+            )
+        }),
+    );
+    let base = start_mock_backend(app).await;
+    let _base_v2 = EnvVarGuard::set("OPENHUMAN_COMPOSIO_DIRECT_BASE_V2", &base);
+    let _base_v3 = EnvVarGuard::set("OPENHUMAN_COMPOSIO_DIRECT_BASE_V3", &base);
+
+    let tmp = tempfile::tempdir().unwrap();
+    let config = direct_mode_no_key_config(&tmp);
+    let _auth_guard = DirectAuthFailureGuard::new("ck_invalid_direct");
+
+    let err = composio_set_api_key(&config, "ck_invalid_direct", false)
+        .await
+        .expect_err("invalid direct-mode key must be rejected before persistence");
+    assert!(
+        err.contains("Invalid Composio API key"),
+        "unexpected error: {err}"
+    );
+    assert_eq!(
+        get_composio_api_key(&config).expect("read composio key after failed save"),
+        None,
+        "invalid key must not be stored"
+    );
+}
+
+#[tokio::test]
+async fn composio_set_api_key_validates_candidate_key_even_when_stored_key_exists() {
+    use crate::openhuman::config::TEST_ENV_LOCK;
+    use crate::openhuman::credentials::{get_composio_api_key, store_composio_api_key};
+    use std::sync::{Arc, Mutex};
+
+    let _env_guard = TEST_ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+    let seen_keys = Arc::new(Mutex::new(Vec::<String>::new()));
+    let app = Router::new()
+        .route(
+            "/connected_accounts",
+            get(
+                |State(seen_keys): State<Arc<Mutex<Vec<String>>>>, headers: HeaderMap| async move {
+                    let key = headers
+                        .get("x-api-key")
+                        .and_then(|value| value.to_str().ok())
+                        .unwrap_or_default()
+                        .to_string();
+                    seen_keys.lock().unwrap().push(key.clone());
+                    if key == "ck_old_valid" {
+                        (axum::http::StatusCode::OK, Json(json!({ "items": [] })))
+                    } else {
+                        (
+                            axum::http::StatusCode::UNAUTHORIZED,
+                            Json(json!({ "error": { "message": "Invalid API key" } })),
+                        )
+                    }
+                },
+            ),
+        )
+        .with_state(seen_keys.clone());
+    let base = start_mock_backend(app).await;
+    let _base_v2 = EnvVarGuard::set("OPENHUMAN_COMPOSIO_DIRECT_BASE_V2", &base);
+    let _base_v3 = EnvVarGuard::set("OPENHUMAN_COMPOSIO_DIRECT_BASE_V3", &base);
+
+    let tmp = tempfile::tempdir().unwrap();
+    let config = direct_mode_no_key_config(&tmp);
+    store_composio_api_key(&config, "ck_old_valid")
+        .await
+        .expect("seed old stored composio key");
+    let _new_key_guard = DirectAuthFailureGuard::new("ck_new_invalid");
+
+    let err = composio_set_api_key(&config, "ck_new_invalid", false)
+        .await
+        .expect_err("candidate invalid key must be rejected even if old stored key is valid");
+    assert!(
+        err.contains("Invalid Composio API key"),
+        "unexpected error: {err}"
+    );
+    assert_eq!(
+        get_composio_api_key(&config).expect("read composio key after failed replacement"),
+        Some("ck_old_valid".to_string()),
+        "failed replacement must leave the old stored key intact"
+    );
+    assert_eq!(
+        seen_keys.lock().unwrap().as_slice(),
+        ["ck_new_invalid"],
+        "validation must probe the candidate key, not the stored key"
+    );
+}
+
 // ── Direct-mode authorize / list_tools / execute (commit 1, #1710) ─
 
 /// Direct-mode `composio_list_tools` now hits Composio v3 with the

src/openhuman/composio/tools/direct.rs

@@ -90,6 +90,10 @@ impl ComposioTool {
         )
     }
 
+    pub(crate) fn auth_key_fingerprint(&self) -> u64 {
+        crate::openhuman::composio::direct_auth::fingerprint_api_key(&self.api_key)
+    }
+
     /// Debug-test seam for raw integration coverage: construct a direct
     /// Composio tool against explicit v2/v3 base URLs. Non-HTTPS URLs are
     /// accepted only for loopback hosts and only in debug builds.

tests/composio_credentials_state_raw_coverage_e2e.rs

@@ -12,7 +12,7 @@ use axum::body::to_bytes;
 use axum::extract::{Request, State};
 use axum::http::{Method, StatusCode};
 use axum::response::{IntoResponse, Response};
-use axum::routing::any;
+use axum::routing::{any, get};
 use axum::{Json, Router};
 use chrono::{Duration as ChronoDuration, Utc};
 use serde_json::{json, Value};
@@ -388,6 +388,13 @@ async fn round15_composio_direct_key_mode_flips_without_network() {
     let _lock = env_lock();
     let harness = setup("http://127.0.0.1:9");
     let config = harness.config().await;
+    let direct_base = start_loopback_backend(Router::new().route(
+        "/connected_accounts",
+        get(composio_direct_connected_accounts),
+    ))
+    .await;
+    let _direct_v2_guard = EnvGuard::set("OPENHUMAN_COMPOSIO_DIRECT_BASE_V2", &direct_base);
+    let _direct_v3_guard = EnvGuard::set("OPENHUMAN_COMPOSIO_DIRECT_BASE_V3", &direct_base);
 
     let empty = composio_set_api_key(&config, "   ", false)
         .await
@@ -850,6 +857,10 @@ async fn composio_backend_handler(State(state): State<MockState>, request: Reque
     }
 }
 
+async fn composio_direct_connected_accounts() -> Json<Value> {
+    Json(json!({ "items": [] }))
+}
+
 async fn start_loopback_backend(app: Router) -> String {
     let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
         .await

tests/composio_ops_raw_coverage_e2e.rs

@@ -4,13 +4,13 @@
 //! public ops layer instead of unit-test-only helpers so coverage lands on the
 //! RPC-facing paths used by controllers, tools, and prompt integration fetches.
 
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, OnceLock};
 
 use axum::body::to_bytes;
 use axum::extract::{Request, State};
 use axum::http::{Method, StatusCode};
 use axum::response::{IntoResponse, Response};
-use axum::routing::any;
+use axum::routing::{any, get};
 use axum::{Json, Router};
 use serde_json::Map;
 use serde_json::{json, Value};
@@ -50,6 +50,37 @@ struct RecordedRequest {
     body: Value,
 }
 
+static COMPOSIO_OPS_ENV_LOCK: OnceLock<Mutex<()>> = OnceLock::new();
+
+struct EnvGuard {
+    key: &'static str,
+    old: Option<String>,
+}
+
+impl EnvGuard {
+    fn set(key: &'static str, value: &str) -> Self {
+        let old = std::env::var(key).ok();
+        std::env::set_var(key, value);
+        Self { key, old }
+    }
+}
+
+impl Drop for EnvGuard {
+    fn drop(&mut self) {
+        match &self.old {
+            Some(value) => std::env::set_var(self.key, value),
+            None => std::env::remove_var(self.key),
+        }
+    }
+}
+
+fn env_lock() -> std::sync::MutexGuard<'static, ()> {
+    COMPOSIO_OPS_ENV_LOCK
+        .get_or_init(|| Mutex::new(()))
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner())
+}
+
 #[tokio::test]
 async fn composio_ops_use_loopback_backend_for_happy_and_error_paths() {
     let state = MockState::default();
@@ -323,6 +354,15 @@ async fn composio_ops_use_loopback_backend_for_happy_and_error_paths() {
 
 #[tokio::test]
 async fn composio_direct_key_ops_and_agent_tools_take_local_validation_paths() {
+    let _lock = env_lock();
+    let direct_base = start_loopback_backend(Router::new().route(
+        "/connected_accounts",
+        get(composio_direct_connected_accounts),
+    ))
+    .await;
+    let _direct_v2_guard = EnvGuard::set("OPENHUMAN_COMPOSIO_DIRECT_BASE_V2", &direct_base);
+    let _direct_v3_guard = EnvGuard::set("OPENHUMAN_COMPOSIO_DIRECT_BASE_V3", &direct_base);
+
     let dir = tempdir().expect("tempdir");
     let mut config = Config {
         workspace_dir: dir.path().join("workspace"),
@@ -727,6 +767,10 @@ async fn start_loopback_backend(app: Router) -> String {
     format!("http://127.0.0.1:{}", addr.port())
 }
 
+async fn composio_direct_connected_accounts() -> Json<Value> {
+    Json(json!({ "items": [] }))
+}
+
 fn store_app_session_token(config: &Config, token: &str) {
     AuthService::from_config(config)
         .store_provider_token(

tests/worker_c_modules_e2e.rs

@@ -5,11 +5,13 @@
 //! an isolated workspace. It avoids live network calls.
 
 use std::path::Path;
-use std::sync::{Mutex, OnceLock};
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex, OnceLock};
 use std::time::Duration;
 
+use axum::extract::State;
 use axum::http::header::{AUTHORIZATION, CONTENT_TYPE};
-use axum::{response::Html, routing::get, Router};
+use axum::{response::Html, routing::get, Json, Router};
 use reqwest::StatusCode;
 use serde_json::{json, Value};
 use tempfile::{tempdir, TempDir};
@@ -124,6 +126,8 @@ async fn setup() -> Harness {
         EnvVarGuard::unset("BACKEND_URL"),
         EnvVarGuard::unset("VITE_BACKEND_URL"),
         EnvVarGuard::unset("OPENHUMAN_API_URL"),
+        EnvVarGuard::unset("OPENHUMAN_COMPOSIO_DIRECT_BASE_V2"),
+        EnvVarGuard::unset("OPENHUMAN_COMPOSIO_DIRECT_BASE_V3"),
         EnvVarGuard::set("OPENHUMAN_KEYRING_BACKEND", "file"),
         EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_STRICT", "false"),
         EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_ENDPOINT", ""),
@@ -362,6 +366,9 @@ async fn channels_remaining_controller_paths_validate_without_live_services() {
 async fn composio_direct_mode_api_key_and_static_catalogs_round_trip() {
     let _lock = env_lock();
     let harness = setup().await;
+    let (composio_base, composio_hits, composio_join) = serve_composio_direct_fixtures().await;
+    let _composio_v2_guard = EnvVarGuard::set("OPENHUMAN_COMPOSIO_DIRECT_BASE_V2", &composio_base);
+    let _composio_v3_guard = EnvVarGuard::set("OPENHUMAN_COMPOSIO_DIRECT_BASE_V3", &composio_base);
 
     let capabilities = rpc(
         &harness.rpc_base,
@@ -430,6 +437,11 @@ async fn composio_direct_mode_api_key_and_static_catalogs_round_trip() {
             .and_then(Value::as_str),
         Some("direct")
     );
+    assert_eq!(
+        composio_hits.load(Ordering::SeqCst),
+        1,
+        "saving a direct-mode API key should validate the candidate key against the v3 mock"
+    );
 
     let mode1 = rpc(
         &harness.rpc_base,
@@ -480,6 +492,7 @@ async fn composio_direct_mode_api_key_and_static_catalogs_round_trip() {
             .and_then(Value::as_str),
         Some("backend")
     );
+    composio_join.abort();
 }
 
 #[tokio::test]
@@ -1005,6 +1018,32 @@ async fn serve_source_fixtures() -> (String, tokio::task::JoinHandle<Result<(),
     (format!("http://{addr}"), join)
 }
 
+async fn serve_composio_direct_fixtures() -> (
+    String,
+    Arc<AtomicUsize>,
+    tokio::task::JoinHandle<Result<(), std::io::Error>>,
+) {
+    async fn connected_accounts(State(hits): State<Arc<AtomicUsize>>) -> Json<Value> {
+        hits.fetch_add(1, Ordering::SeqCst);
+        Json(json!({
+            "items": []
+        }))
+    }
+
+    let hits = Arc::new(AtomicUsize::new(0));
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("bind composio fixture listener");
+    let addr = listener
+        .local_addr()
+        .expect("composio fixture listener addr");
+    let app = Router::new()
+        .route("/connected_accounts", get(connected_accounts))
+        .with_state(hits.clone());
+    let join = tokio::spawn(async move { axum::serve(listener, app).await });
+    (format!("http://{addr}"), hits, join)
+}
+
 #[tokio::test]
 async fn memory_sources_folder_web_and_rss_readers_sync_through_rpc() {
     let _lock = env_lock();