fix(observability,composio): demote direct-mode Composio 401/Invalid API key noise (Sentry TAURI-RUST-X9) (#2481)

Co-authored-by: Steven Enamakel <enamakel@tinyhumans.ai>

src/core/observability.rs

@@ -518,6 +518,42 @@ fn is_provider_user_state_message(lower: &str) -> bool {
         return true;
     }
 
+    // TAURI-RUST-X9 (#1166): direct-mode composio call against the user's
+    // personal Composio v3 tenant rejected with a 401 because the stored
+    // API key is invalid / revoked / has the wrong prefix. The canonical
+    // wire shape rendered by
+    // `src/openhuman/composio/tools/impl/network/composio.rs::response_error`
+    // and the various direct-mode op wrappers is:
+    //
+    //   `[composio-direct] list_connections failed: Composio v3
+    //    connected_accounts failed: HTTP 401: Invalid API key: ak_…`
+    //
+    // The "Invalid API key" body is rendered for every direct-mode
+    // endpoint (list_connections / list_tools / authorize / etc.), so we
+    // gate on the **`[composio-direct]` prefix** + either of the two
+    // anchors that prove the failure came from the v3 auth wall:
+    //   - `HTTP 401`  (the status the v3 wall returns)
+    //   - `Invalid API key`  (the body Composio puts in the JSON)
+    //
+    // Requiring the `[composio-direct]` prefix keeps this from
+    // accidentally swallowing unrelated bugs — backend-mode 401s from
+    // `integrations/composio/*` still carry the `Backend returned 401`
+    // shape (handled by the failure-tag flow with `status="401"`),
+    // not the `HTTP 401: Invalid API key` shape.
+    //
+    // Remediation is purely user-state: the user must rotate / re-enter
+    // their Composio key via Settings → Composio → Direct mode. Sentry
+    // has no actionable signal — the UI surfaces the "Invalid API key"
+    // toast and the polling layer already retries every 5 s.
+    //
+    // Drops Sentry TAURI-RUST-X9 (~15.7 k events / ~22 h, single user,
+    // release openhuman@0.54.0+c25fc8e5fd3e).
+    if lower.contains("[composio-direct]")
+        && (lower.contains("http 401") || lower.contains("invalid api key"))
+    {
+        return true;
+    }
+
     false
 }
 
@@ -2004,6 +2040,118 @@ mod tests {
         );
     }
 
+    // ── TAURI-RUST-X9 (#1166): composio-direct 401 / Invalid API key ────
+
+    #[test]
+    fn classifies_composio_direct_invalid_api_key_as_provider_user_state() {
+        // Canonical Sentry TAURI-RUST-X9 wire shape — the verbatim title
+        // body from the issue, captured 15,732 times in ~22h on a single
+        // user with a bad direct-mode key. The classifier must demote
+        // this to `ProviderUserState` so the polling layer's 5 s retry
+        // doesn't keep flooding Sentry.
+        let msg = "[composio-direct] list_connections failed: \
+                   Composio v3 connected_accounts failed: \
+                   HTTP 401: Invalid API key: ak_VsUvq*****";
+        assert_eq!(
+            expected_error_kind(msg),
+            Some(ExpectedErrorKind::ProviderUserState),
+            "composio-direct HTTP 401 + Invalid API key must demote to ProviderUserState"
+        );
+    }
+
+    #[test]
+    fn classifies_composio_direct_invalid_api_key_for_other_ops() {
+        // Same arm must cover every op-name the direct branches emit —
+        // not just `list_connections`. The matcher gates on the
+        // `[composio-direct]` prefix, not on a specific op string, so
+        // `list_tools` / `authorize` / `list_connections` all demote.
+        let shapes = [
+            // list_tools prefetch fails before the actual list_tools call
+            "[composio-direct] list_tools: prefetch connections failed: \
+             Composio v3 connected_accounts failed: HTTP 401: Invalid API key: ak_…",
+            // direct authorize hits the v3 /connected_accounts/link wall
+            "[composio-direct] authorize failed: \
+             Composio v3 connected_accounts/link failed: HTTP 401: Invalid API key: ak_…",
+            // direct list_tools itself
+            "[composio-direct] list_tools failed: \
+             Composio v3 tools failed: HTTP 401: Invalid API key: ak_…",
+            // periodic-tick rendering (no "[composio-direct]" prefix because
+            // periodic.rs wraps differently, but the failure still gets the
+            // hook — handled by ops.rs's report path, not the
+            // expected_error_kind body shape, so we only verify the
+            // composio-direct branch here)
+        ];
+        for msg in shapes {
+            assert_eq!(
+                expected_error_kind(msg),
+                Some(ExpectedErrorKind::ProviderUserState),
+                "every [composio-direct] op with HTTP 401 / Invalid API key must demote: {msg}"
+            );
+        }
+    }
+
+    #[test]
+    fn classifies_composio_direct_with_invalid_api_key_only_no_http_401() {
+        // The matcher accepts EITHER `HTTP 401` OR `Invalid API key`
+        // alongside the `[composio-direct]` prefix. Catches the wire
+        // shape variant where the body anchor lands but the status text
+        // is rendered differently (e.g. "401 Unauthorized" instead of
+        // "HTTP 401") — same user-state condition.
+        let msg = "[composio-direct] list_connections failed: \
+                   Composio v3 connected_accounts failed: \
+                   401 Unauthorized: Invalid API key: ak_…";
+        assert_eq!(
+            expected_error_kind(msg),
+            Some(ExpectedErrorKind::ProviderUserState),
+            "composio-direct + Invalid API key body must demote even without literal 'HTTP 401'"
+        );
+    }
+
+    #[test]
+    fn does_not_classify_unrelated_http_401_as_composio_direct_user_state() {
+        // Discrimination test: a generic 401 that does NOT carry the
+        // `[composio-direct]` prefix must NOT match this arm. This
+        // protects against the arm accidentally swallowing backend-mode
+        // composio 401s, unrelated integration 401s, or any other
+        // 401-containing message that lacks the direct-mode anchor.
+        //
+        // The backend-mode shape is `Backend returned 401 …`; it does
+        // not contain `[composio-direct]`, so the new arm rightly skips
+        // it. Backend-mode 401s remain a real Sentry signal (bad
+        // service-to-service auth, expired token, etc.).
+        let backend_401 = "[composio] list_connections failed: \
+                           Backend returned 401 Unauthorized for GET \
+                           https://api.tinyhumans.ai/agent-integrations/composio/connections: \
+                           Invalid API key";
+        assert_ne!(
+            expected_error_kind(backend_401),
+            Some(ExpectedErrorKind::ProviderUserState),
+            "backend-mode 401 must NOT demote via the composio-direct arm"
+        );
+
+        let unrelated_401 = "GitHub API error: HTTP 401: Bad credentials";
+        assert_ne!(
+            expected_error_kind(unrelated_401),
+            Some(ExpectedErrorKind::ProviderUserState),
+            "unrelated 401 (no [composio-direct] anchor) must NOT match the composio-direct arm"
+        );
+    }
+
+    #[test]
+    fn does_not_classify_composio_direct_500_as_user_state() {
+        // Real bug shapes — a 500 from the direct v3 path with no auth
+        // body anchor — must still fall through to `None` so Sentry
+        // sees them. Without this guard the arm could be too permissive
+        // and silence genuine backend faults.
+        let msg = "[composio-direct] list_connections failed: \
+                   Composio v3 connected_accounts failed: HTTP 500";
+        assert_eq!(
+            expected_error_kind(msg),
+            None,
+            "composio-direct 500 with no auth body must NOT demote — it is a real bug shape"
+        );
+    }
+
     #[test]
     fn classifies_local_ai_binary_missing_errors() {
         // OPENHUMAN-TAURI-9N: `local_ai_tts` returns this exact string

src/openhuman/composio/ops.rs

@@ -87,7 +87,7 @@ fn resolve_client(config: &Config) -> OpResult<ComposioClient> {
 /// handshake eof`, …), we tag `failure="transport"` instead so the
 /// `before_send` filter's transport-phrase branch fires — and keep the
 /// status tag absent (transport failures don't carry a status).
-fn report_composio_op_error<E: std::fmt::Display + ?Sized>(operation: &str, err: &E) {
+pub(super) fn report_composio_op_error<E: std::fmt::Display + ?Sized>(operation: &str, err: &E) {
     // `{err:#}` renders the full anyhow chain when applicable; for plain
     // `String` / `&str` errors it falls back to the Display impl.
     let rendered = format!("{err:#}");
@@ -243,9 +243,22 @@ pub async fn composio_list_connections(
                 "[composio-direct] list_connections: fetching v3 \
                  /connected_accounts for the user's personal Composio tenant"
             );
-            let resp = direct_list_connections(&direct)
-                .await
-                .map_err(|e| format!("[composio-direct] list_connections failed: {e:#}"))?;
+            let resp = direct_list_connections(&direct).await.map_err(|e| {
+                // [#1166 / Sentry TAURI-RUST-X9] Restore symmetric error
+                // routing for the direct-mode branch. Without this hook the
+                // direct-mode 401 ("Invalid API key …") wire shape bypassed
+                // `report_error_or_expected` and leaked ~15.7k events in ~22h
+                // — same UI 5 s poll + `periodic.rs` tick that the
+                // backend branch (line ~266) was already classifying.
+                //
+                // Render WITH the `[composio-direct]` anchor BEFORE
+                // reporting so the classifier arm in
+                // `is_provider_user_state_message` (which gates on that
+                // prefix) actually fires.
+                let rendered = format!("[composio-direct] list_connections failed: {e:#}");
+                report_composio_op_error("list_connections", &rendered);
+                rendered
+            })?;
             let active = resp.connections.iter().filter(|c| c.is_active()).count();
             let total = resp.connections.len();
             // Reconcile the integrations cache against this fresh live
@@ -338,7 +351,16 @@ pub async fn composio_authorize(
             .await
             .map_err(|e| {
                 let wrapped = super::oauth_handoff::wrap_authorize_rate_limit_error(toolkit, e);
-                format!("[composio-direct] authorize failed: {wrapped:#}")
+                // [#1166 / Sentry TAURI-RUST-X9] Symmetric with the
+                // backend branch's `report_composio_op_error` on the
+                // same handler — direct-mode 401s from
+                // `connected_accounts/link` were leaking otherwise.
+                // Render WITH the `[composio-direct]` anchor so the
+                // classifier arm fires; wrapped error preserves any
+                // rate-limit classifications fed up the ladder.
+                let rendered = format!("[composio-direct] authorize failed: {wrapped:#}");
+                report_composio_op_error("authorize", &rendered);
+                rendered
             })?
         }
     };
@@ -480,7 +502,17 @@ pub async fn composio_list_tools(
                 Some(list) if !list.is_empty() => list,
                 _ => {
                     let conns = direct_list_connections(&direct).await.map_err(|e| {
-                        format!("[composio-direct] list_tools: prefetch connections failed: {e:#}")
+                        // [#1166 / Sentry TAURI-RUST-X9] Symmetric error
+                        // routing — the prefetch call goes to the same v3
+                        // `/connected_accounts` endpoint as `list_connections`
+                        // and would emit the same 401 wire shape. Render
+                        // WITH the `[composio-direct]` anchor so the
+                        // classifier arm fires on the prefetch path too.
+                        let rendered = format!(
+                            "[composio-direct] list_tools: prefetch connections failed: {e:#}"
+                        );
+                        report_composio_op_error("list_connections", &rendered);
+                        rendered
                     })?;
                     let mut v: Vec<String> = conns
                         .connections
@@ -510,9 +542,16 @@ pub async fn composio_list_tools(
                 toolkits = scope.len(),
                 "[composio-direct] list_tools: fetching v3 tool schemas"
             );
-            let mut resp = direct_list_tools(&direct, &scope)
-                .await
-                .map_err(|e| format!("[composio-direct] list_tools failed: {e:#}"))?;
+            let mut resp = direct_list_tools(&direct, &scope).await.map_err(|e| {
+                // [#1166 / Sentry TAURI-RUST-X9] Symmetric with the backend
+                // branch's hook (line ~451). Direct-mode `list_tools`
+                // failures are user-state when the API key is bad. Render
+                // WITH the `[composio-direct]` anchor so the classifier
+                // arm fires.
+                let rendered = format!("[composio-direct] list_tools failed: {e:#}");
+                report_composio_op_error("list_tools", &rendered);
+                rendered
+            })?;
             // Apply the same curated-whitelist + user-scope filter the
             // backend path runs — schemas may be tenant-agnostic but
             // OpenHuman's curation policy isn't, and direct-mode users

src/openhuman/composio/ops_test.rs

@@ -1619,3 +1619,85 @@ fn composio_transport_timeout_is_dropped_by_before_send() {
         "composio transport timeout must be dropped by integrations filter (#1608)"
     );
 }
+
+// ── TAURI-RUST-X9 (#1166): direct-mode auth-rejection routing ───────────
+//
+// Pins the contract that direct-mode 401 / Invalid API key shapes are
+// classified by the observability matcher AND their failure-tag stays
+// `non_2xx` so the `before_send` integrations filter has consistent
+// inputs. Together with the classifier-arm tests in
+// `core::observability` these tests prove the leak path (~15.7 k events
+// in ~22h before #1166) is closed end-to-end.
+
+#[test]
+fn composio_direct_invalid_api_key_classifies_as_provider_user_state() {
+    // The verbatim Sentry TAURI-RUST-X9 wire shape — emitted by
+    // `ops.rs::composio_list_connections` direct branch via the
+    // `report_composio_op_error` hook restored in #1166. Routing this
+    // through `expected_error_kind` is what demotes it to
+    // `ProviderUserState` (info breadcrumb) instead of firing a Sentry
+    // event.
+    let msg = "[composio-direct] list_connections failed: \
+               Composio v3 connected_accounts failed: \
+               HTTP 401: Invalid API key: ak_VsUvq*****";
+    assert_eq!(
+        crate::core::observability::expected_error_kind(msg),
+        Some(crate::core::observability::ExpectedErrorKind::ProviderUserState),
+        "the canonical TAURI-RUST-X9 wire shape must demote via the composio-direct arm"
+    );
+}
+
+#[test]
+fn composio_direct_invalid_api_key_failure_tag_is_non_2xx() {
+    // Belt-and-suspenders: even if `expected_error_kind` ever stops
+    // matching the body (regression in the classifier arm), the
+    // failure tag must STILL be `non_2xx`. Combined with the
+    // `before_send` filter's transient-status handling and a
+    // future-added `status="401"` tag (Patch 1 doesn't extract status
+    // from the `HTTP 401` shape — only the `Backend returned <status>`
+    // shape — so this just pins the safe default), this is the
+    // backstop drop path.
+    let rendered = "[composio-direct] list_connections failed: \
+                    Composio v3 connected_accounts failed: \
+                    HTTP 401: Invalid API key: ak_VsUvq*****";
+    assert_eq!(
+        classify_composio_failure_tag(rendered),
+        "non_2xx",
+        "direct-mode auth-rejection must tag as non_2xx (not transport)"
+    );
+}
+
+#[test]
+fn composio_direct_invalid_api_key_extract_status_returns_none() {
+    // Pins the contract: `extract_backend_returned_status` only parses
+    // the integrations-layer `Backend returned <status>` rendering, NOT
+    // the direct-mode `HTTP 401` shape. The direct-mode arm relies on
+    // the classifier demotion + the failure-tag drop path instead of
+    // status extraction; if this ever changes (e.g. we extend the
+    // status extractor to cover both shapes), the new behaviour should
+    // come with an explicit test, not be inferred.
+    let rendered = "[composio-direct] list_connections failed: \
+                    Composio v3 connected_accounts failed: \
+                    HTTP 401: Invalid API key: ak_…";
+    assert_eq!(
+        extract_backend_returned_status(rendered),
+        None,
+        "direct-mode HTTP 401 must not parse via extract_backend_returned_status"
+    );
+}
+
+#[test]
+fn composio_direct_500_does_not_demote() {
+    // Discrimination test from the composio side — a real bug shape
+    // (500 with no auth body) MUST escape the classifier and reach
+    // `report_error_message`. Without this guard the matcher in
+    // `observability.rs` could be tightened too far and silence
+    // genuine backend faults.
+    let msg = "[composio-direct] list_connections failed: \
+               Composio v3 connected_accounts failed: HTTP 500";
+    assert_eq!(
+        crate::core::observability::expected_error_kind(msg),
+        None,
+        "composio-direct 500 with no auth body must remain an unclassified bug shape"
+    );
+}

src/openhuman/composio/periodic.rs

@@ -175,9 +175,22 @@ pub(crate) async fn run_one_tick() -> Result<(), String> {
             .list_connections()
             .await
             .map_err(|e| format!("list_connections (backend): {e}"))?,
-        ComposioClientKind::Direct(direct) => direct_list_connections(direct)
-            .await
-            .map_err(|e| format!("list_connections (direct): {e:#}"))?,
+        ComposioClientKind::Direct(direct) => {
+            direct_list_connections(direct).await.map_err(|e| {
+                // [#1166 / Sentry TAURI-RUST-X9] The server-side periodic
+                // tick re-renders the same v3 `/connected_accounts` 401
+                // shape that `ops::composio_list_connections` emits, so
+                // route it through the observability classifier too.
+                // Without this, the tick-side 401s leak as unclassified
+                // Sentry events even when the UI poll's identical failure
+                // is correctly classified. Render WITH the
+                // `[composio-direct]` anchor so the classifier arm in
+                // `is_provider_user_state_message` actually fires.
+                let rendered = format!("[composio-direct] list_connections (direct): {e:#}");
+                super::ops::report_composio_op_error("list_connections", &rendered);
+                rendered
+            })?
+        }
     };
 
     let sync_map = last_sync_map();

src/openhuman/composio/tools.rs

@@ -504,7 +504,20 @@ impl Tool for ComposioListConnectionsTool {
             Ok(ComposioClientKind::Direct(direct)) => {
                 tracing::debug!("[composio-direct] list_connections.execute: direct variant");
                 direct_list_connections(&direct).await.map_err(|e| {
-                    anyhow::anyhow!("composio_list_connections (direct) failed: {e}")
+                    // [#1166 / Sentry TAURI-RUST-X9] Symmetric error
+                    // routing with `ops.rs::composio_list_connections`.
+                    // The agent-tool path can also fire 401s when a
+                    // direct-mode user has a bad API key — without this
+                    // hook the failure escapes the classifier and lands
+                    // as an unclassified Sentry event. Render WITH the
+                    // `[composio-direct]` anchor BEFORE reporting so the
+                    // classifier arm in `is_provider_user_state_message`
+                    // (gated on that prefix) actually fires.
+                    let rendered = format!(
+                        "[composio-direct] composio_list_connections (direct) failed: {e:#}"
+                    );
+                    super::ops::report_composio_op_error("list_connections", &rendered);
+                    anyhow::anyhow!("{rendered}")
                 })?
             }
             Err(e) => {