admin/openhuman
1
0
Fork 0
mirror of https://github.com/tinyhumansai/openhuman.git synced 2026-07-27 21:08:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

composio: preserve Composio tool error semantics instead of bucketing as 502 (#1797) (#1827)

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Steven Enamakel <enamakel@tinyhumans.ai>
This commit is contained in:
CodeGhost21
2026-05-15 20:59:52 -07:00
committed by GitHub
co-authored by Cursor Steven Enamakel
parent bbc5504691
commit 258837e17b
16 changed files with 900 additions and 222 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/src/lib/composio/formatters.test.ts
+54 -1
View File
@@ -1,6 +1,59 @@
import { describe, expect, it } from 'vitest';
import { formatTriggerLabel } from './formatters';
import { formatComposioToolError, formatTriggerLabel } from './formatters';
describe('formatComposioToolError', () => {
it('strips the classified prefix and returns the body', () => {
const raw =
'[composio:error:insufficient_scope] `GMAIL_FETCH_EMAILS` was rejected because the connected gmail account is missing required permissions.';
expect(formatComposioToolError(raw)).toContain('missing required permissions');
expect(formatComposioToolError(raw)).not.toContain('[composio:error:');
});
it('passes through unclassified messages', () => {
expect(formatComposioToolError('plain failure')).toBe('plain failure');
});
it('returns empty string for null/undefined/empty input', () => {
expect(formatComposioToolError(null)).toBe('');
expect(formatComposioToolError(undefined)).toBe('');
expect(formatComposioToolError('')).toBe('');
});
it('falls back to validation copy when body is empty', () => {
expect(formatComposioToolError('[composio:error:validation]')).toBe('Invalid tool arguments.');
});
it('falls back to insufficient_scope copy when body is empty', () => {
expect(formatComposioToolError('[composio:error:insufficient_scope]')).toBe(
'Reconnect this integration and grant the requested permissions.'
);
});
it('falls back to rate_limited copy when body is empty', () => {
expect(formatComposioToolError('[composio:error:rate_limited]')).toBe(
'The upstream service is rate-limiting requests. Try again shortly.'
);
});
it('falls back to gateway copy when body is empty', () => {
expect(formatComposioToolError('[composio:error:gateway]')).toBe(
'Temporary connection issue. Try again in a moment.'
);
});
it('returns body for unknown class when present', () => {
expect(formatComposioToolError('[composio:error:something_new] details here')).toBe(
'details here'
);
});
it('falls back to trimmed raw for unknown class with empty body', () => {
expect(formatComposioToolError(' [composio:error:something_new] ')).toBe(
'[composio:error:something_new]'
);
});
});
describe('formatTriggerLabel', () => {
it('formats GOOGLECALENDAR_GOOGLE_CALENDAR_EVENT_CREATED_TRIGGER correctly', () => {
app/src/lib/composio/formatters.ts
+23
View File
@@ -11,6 +11,29 @@
* 4. dedupe leading provider prefix when it reappears
* 5. split on _, title-case each token, join with space
*/
/**
* Parse a classified Composio error (`[composio:error:<class>] …`) for UI copy.
*/
export function formatComposioToolError(raw: string | null | undefined): string {
if (!raw) return '';
const match = /^\[composio:error:([a-z_]+)\]\s*(.*)$/is.exec(raw.trim());
if (!match) return raw.trim();
const [, className, body] = match;
switch (className) {
case 'validation':
return body || 'Invalid tool arguments.';
case 'insufficient_scope':
return body || 'Reconnect this integration and grant the requested permissions.';
case 'rate_limited':
return body || 'The upstream service is rate-limiting requests. Try again shortly.';
case 'gateway':
return body || 'Temporary connection issue. Try again in a moment.';
default:
return body || raw.trim();
}
}
export function formatTriggerLabel(
slug: string | null | undefined,
opts?: { overrides?: Record<string, string> }
src/openhuman/composio/action_tool.rs
+14 -29
View File
@@ -29,7 +29,7 @@ use std::sync::Arc;
use async_trait::async_trait;
use serde_json::Value;
use super::client::{create_composio_client, direct_execute, ComposioClientKind};
use super::client::create_composio_client;
use super::providers::ToolScope;
use super::tools::resolve_action_scope;
use crate::openhuman::agent::harness::current_sandbox_mode;
@@ -195,33 +195,18 @@ impl Tool for ComposioActionTool {
};
let started = std::time::Instant::now();
let res = match kind {
ComposioClientKind::Backend(client) => {
tracing::debug!(
tool = %self.action_name,
"[composio] per-action execute: backend variant"
);
// Wrap with auth_retry so a stale tinyhumans-tenant
// JWT gets refreshed-and-replayed once before surfacing
// (upstream behaviour).
super::auth_retry::execute_with_auth_retry(&client, &self.action_name, args).await
}
ComposioClientKind::Direct(direct) => {
tracing::debug!(
tool = %self.action_name,
"[composio] per-action execute: direct variant"
);
// Direct path skips auth_retry — see ComposioExecuteTool
// for rationale (no backend refresh surface).
direct_execute(
&direct,
&self.action_name,
args,
&live_config.composio.entity_id,
)
.await
}
};
// Route through the centralized dispatcher (#1797) so both
// backend and direct variants share the same prepare/retry/error-
// mapping pipeline. The dispatcher applies `format_provider_error`
// to failures (transport + provider) so downstream consumers can
// parse `[composio:error:<class>] …`.
let res = super::execute_dispatch::execute_composio_action_kind(
kind,
&self.action_name,
args,
&live_config.composio.entity_id,
)
.await;
let elapsed_ms = started.elapsed().as_millis() as u64;
match res {
@@ -267,7 +252,7 @@ impl Tool for ComposioActionTool {
elapsed_ms,
},
);
Ok(ToolResult::error(format!("{}: {e}", self.action_name)))
Ok(ToolResult::error(e))
}
}
}
src/openhuman/composio/client.rs
+26 -58
View File
@@ -166,7 +166,8 @@ impl ComposioClient {
if tool.is_empty() {
anyhow::bail!("composio.execute_tool_once: tool slug must not be empty");
}
let arguments = arguments.unwrap_or(serde_json::Value::Object(Default::default()));
let arguments = super::execute_prepare::prepare_execute_arguments(tool, arguments)
.map_err(anyhow::Error::msg)?;
tracing::debug!(tool = %tool, "[composio] execute_tool_once (no built-in retry)");
let body = json!({ "tool": tool, "arguments": arguments });
let result = self.post_execute_tool(&body).await;
@@ -183,7 +184,12 @@ impl ComposioClient {
"[composio] execute_tool_once failed"
),
}
result
result.map_err(|e| {
anyhow::Error::msg(super::error_mapping::remap_transport_error(
tool,
&e.to_string(),
))
})
}
/// `POST /agent-integrations/composio/execute` — run a Composio
@@ -197,12 +203,26 @@ impl ComposioClient {
if tool.is_empty() {
anyhow::bail!("composio.execute_tool: tool slug must not be empty");
}
let mut arguments = arguments.unwrap_or(serde_json::Value::Object(Default::default()));
normalize_calendar_query_args(tool, &mut arguments);
// PR #1827 routes all execute-side argument normalization
// (including the bare-date → RFC 3339 fix #1802 brought to
// `normalize_calendar_query_args` on `main`) through the
// centralized `prepare_execute_arguments` helper. The helper
// covers the same calendar query case and is the shared entry
// point for `composio_execute`, per-action tools, and direct-
// mode dispatch.
let arguments = super::execute_prepare::prepare_execute_arguments(tool, arguments)
.map_err(anyhow::Error::msg)?;
tracing::debug!(tool = %tool, "[composio] execute_tool");
let body = json!({ "tool": tool, "arguments": arguments });
self.execute_tool_with_post_oauth_retry(tool, &body, POST_OAUTH_ACTION_RETRY_DELAY)
.await
let mut resp = self
.execute_tool_with_post_oauth_retry(tool, &body, POST_OAUTH_ACTION_RETRY_DELAY)
.await?;
if !resp.successful {
if let Some(ref err) = resp.error {
resp.error = Some(super::error_mapping::format_provider_error(tool, err));
}
}
Ok(resp)
}
pub(super) async fn execute_tool_with_post_oauth_retry(
@@ -506,58 +526,6 @@ impl ComposioClient {
}
}
/// Calendar query slugs whose `timeMin`/`timeMax` values should be
/// normalized to RFC 3339 timestamps. LLM-generated arguments sometimes
/// emit bare dates like `"2026-05-14"` instead of
/// `"2026-05-14T00:00:00Z"`, which Google Calendar rejects.
const CALENDAR_QUERY_SLUGS: &[&str] = &["GOOGLECALENDAR_EVENTS_LIST", "GOOGLECALENDAR_FIND_EVENT"];
/// Normalize `timeMin`/`timeMax` from bare dates to RFC 3339 for
/// Google Calendar query slugs. The LLM prompt instructs the model to
/// use RFC 3339 format, but some model invocations still produce bare
/// `YYYY-MM-DD` strings.
fn normalize_calendar_query_args(tool: &str, arguments: &mut serde_json::Value) {
if !CALENDAR_QUERY_SLUGS.contains(&tool) {
return;
}
let Some(map) = arguments.as_object_mut() else {
return;
};
for key in &["timeMin", "timeMax"] {
if let Some(serde_json::Value::String(val)) = map.get(*key).cloned() {
if is_bare_date(&val) {
let normalized = format!("{}T00:00:00Z", val);
tracing::debug!(
tool = %tool,
key = %key,
normalized = %normalized,
"[composio] normalized bare date to RFC 3339 for calendar query"
);
map.insert((*key).to_string(), serde_json::Value::String(normalized));
}
}
}
}
/// Returns `true` when `s` is a bare date string like `"2026-05-14"`
/// with no time component.
fn is_bare_date(s: &str) -> bool {
if s.len() != 10 {
return false;
}
let bytes = s.as_bytes();
bytes[0].is_ascii_digit()
&& bytes[1].is_ascii_digit()
&& bytes[2].is_ascii_digit()
&& bytes[3].is_ascii_digit()
&& bytes[4] == b'-'
&& bytes[5].is_ascii_digit()
&& bytes[6].is_ascii_digit()
&& bytes[7] == b'-'
&& bytes[8].is_ascii_digit()
&& bytes[9].is_ascii_digit()
}
fn is_post_oauth_auth_readiness_error(resp: &ComposioExecuteResponse) -> bool {
if resp.successful {
return false;
src/openhuman/composio/client_tests.rs
+8 -98
View File
@@ -750,7 +750,9 @@ async fn execute_tool_surfaces_non_successful_provider_response() {
);
let base = start_mock_backend(app).await;
let client = build_client_for(base);
let resp = client.execute_tool("GMAIL_SEND_EMAIL", None).await.unwrap();
// Use a slug that bypasses local arg validation in `execute_prepare`
// so the test exercises the gateway-response path, not the pre-flight.
let resp = client.execute_tool("ANY_TOOL", None).await.unwrap();
assert!(
!resp.successful,
"non-successful provider response must be surfaced via the successful flag"
@@ -854,103 +856,11 @@ async fn execute_tool_sends_tool_slug_in_request_body() {
"tool slug must be forwarded in request body"
);
}
// ── Calendar query argument normalization ───────────────────────
#[test]
fn is_bare_date_rejects_non_date_strings() {
assert!(!is_bare_date(""));
assert!(!is_bare_date("2026-05"));
assert!(!is_bare_date("2026-05-14T00:00:00Z"));
assert!(!is_bare_date("2026/05/14"));
assert!(!is_bare_date("hello-world"));
assert!(!is_bare_date("2026-5-14"));
assert!(!is_bare_date("2026-05-144"));
}
#[test]
fn is_bare_date_accepts_valid_date_strings() {
assert!(is_bare_date("2026-05-14"));
assert!(is_bare_date("2025-01-01"));
assert!(is_bare_date("1999-12-31"));
assert!(is_bare_date("0001-01-01"));
}
#[test]
fn normalize_calendar_query_args_ignores_non_calendar_slugs() {
let mut args = serde_json::json!({ "timeMin": "2026-05-14" });
normalize_calendar_query_args("GMAIL_SEND_EMAIL", &mut args);
assert_eq!(args["timeMin"], "2026-05-14");
}
#[test]
fn normalize_calendar_query_args_converts_bare_date_to_rfc3339() {
let mut args = serde_json::json!({
"connectionId": "conn-1",
"timeMin": "2026-05-14",
"timeMax": "2026-05-15",
});
normalize_calendar_query_args("GOOGLECALENDAR_EVENTS_LIST", &mut args);
assert_eq!(args["timeMin"], "2026-05-14T00:00:00Z");
assert_eq!(args["timeMax"], "2026-05-15T00:00:00Z");
assert_eq!(args["connectionId"], "conn-1");
}
#[test]
fn normalize_calendar_query_args_preserves_rfc3339_timestamp() {
let mut args = serde_json::json!({
"timeMin": "2026-05-14T00:00:00+05:30",
"timeMax": "2026-05-14T23:59:59Z",
});
normalize_calendar_query_args("GOOGLECALENDAR_EVENTS_LIST", &mut args);
assert_eq!(args["timeMin"], "2026-05-14T00:00:00+05:30");
assert_eq!(args["timeMax"], "2026-05-14T23:59:59Z");
}
#[test]
fn normalize_calendar_query_args_handles_missing_time_fields() {
let mut args = serde_json::json!({ "connectionId": "conn-1" });
normalize_calendar_query_args("GOOGLECALENDAR_EVENTS_LIST", &mut args);
assert_eq!(args["connectionId"], "conn-1");
// timeMin/timeMax should not be inserted if absent
assert!(args.get("timeMin").is_none());
}
#[test]
fn normalize_calendar_query_args_handles_non_object_arguments() {
let mut args = serde_json::json!("just a string");
normalize_calendar_query_args("GOOGLECALENDAR_EVENTS_LIST", &mut args);
assert_eq!(args, "just a string");
}
#[test]
fn normalize_calendar_query_args_handles_calendar_find_event_slug() {
let mut args = serde_json::json!({ "timeMin": "2026-06-01" });
normalize_calendar_query_args("GOOGLECALENDAR_FIND_EVENT", &mut args);
assert_eq!(args["timeMin"], "2026-06-01T00:00:00Z");
}
#[test]
fn normalize_calendar_query_args_normalizes_one_side_when_other_absent() {
// Asymmetric: only timeMin present, timeMax missing. The normalizer
// must convert timeMin without inserting a synthetic timeMax.
let mut args = serde_json::json!({ "timeMin": "2026-05-14" });
normalize_calendar_query_args("GOOGLECALENDAR_EVENTS_LIST", &mut args);
assert_eq!(args["timeMin"], "2026-05-14T00:00:00Z");
assert!(args.get("timeMax").is_none());
}
#[test]
fn normalize_calendar_query_args_skips_non_string_values() {
// Non-string values (numbers, bools, nulls, objects) must be left
// untouched — the normalizer only rewrites string bare-date inputs.
let mut args = serde_json::json!({
"timeMin": 42,
"timeMax": null,
});
normalize_calendar_query_args("GOOGLECALENDAR_EVENTS_LIST", &mut args);
assert_eq!(args["timeMin"], 42);
assert!(args["timeMax"].is_null());
}
// Calendar bare-date → RFC 3339 normalization is now covered by
// `execute_prepare::prepare_execute_arguments` (PR #1827); see
// `execute_prepare_tests.rs` for the equivalent test surface that
// supersedes the per-slug `normalize_calendar_query_args` helper
// removed alongside the upstream-main merge.
// ── Factory tests (`create_composio_client`) ────────────────────────
//
src/openhuman/composio/error_mapping.rs
+214
View File
@@ -0,0 +1,214 @@
//! Classify and format Composio tool failures so validation, scope, and
//! upstream-provider errors are not surfaced as generic gateway (502) failures.
//!
//! Issue #1797 — Composio support found tool-level failures on their side while
//! OpenHuman was bucketing them as HTTP 502 / gateway instability.
/// Stable, grep-friendly error classes for metrics and UI routing.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ComposioErrorClass {
Validation,
InsufficientScope,
RateLimited,
UpstreamProvider,
ComposioPlatform,
Gateway,
Other,
}
impl ComposioErrorClass {
pub fn as_str(self) -> &'static str {
match self {
Self::Validation => "validation",
Self::InsufficientScope => "insufficient_scope",
Self::RateLimited => "rate_limited",
Self::UpstreamProvider => "upstream_provider",
Self::ComposioPlatform => "composio_platform",
Self::Gateway => "gateway",
Self::Other => "other",
}
}
}
pub fn classify_composio_error(tool: &str, message: &str) -> ComposioErrorClass {
let lower = message.to_ascii_lowercase();
let class = if is_validation_shape(&lower) {
ComposioErrorClass::Validation
} else if is_insufficient_scope_shape(&lower) {
ComposioErrorClass::InsufficientScope
} else if is_rate_limited_shape(&lower) {
ComposioErrorClass::RateLimited
} else if is_gateway_transport_shape(&lower) && !is_embedded_provider_failure(&lower) {
ComposioErrorClass::Gateway
} else if is_composio_platform_shape(&lower) {
ComposioErrorClass::ComposioPlatform
} else if tool.starts_with("GMAIL_")
|| tool.starts_with("SLACK_")
|| tool.starts_with("NOTION_")
|| tool.starts_with("GOOGLECALENDAR_")
{
ComposioErrorClass::UpstreamProvider
} else {
ComposioErrorClass::Other
};
tracing::debug!(
tool = %tool,
class = class.as_str(),
"[composio][classify] error classified"
);
class
}
pub fn format_provider_error(tool: &str, raw: &str) -> String {
let class = classify_composio_error(tool, raw);
let detail = raw.trim();
let body = match class {
ComposioErrorClass::Validation => format!("Invalid arguments for `{tool}`: {detail}"),
ComposioErrorClass::InsufficientScope => format_insufficient_scope_message(tool, detail),
ComposioErrorClass::RateLimited => format_rate_limited_message(tool, detail),
ComposioErrorClass::UpstreamProvider => {
format!("`{tool}` failed at the connected provider: {detail}")
}
ComposioErrorClass::ComposioPlatform => {
format!("Composio connection issue for `{tool}`: {detail}")
}
ComposioErrorClass::Gateway => {
format!("Temporary gateway error while calling `{tool}`: {detail}")
}
ComposioErrorClass::Other => format!("`{tool}` failed: {detail}"),
};
prefix_class(class, &body)
}
pub fn remap_transport_error(tool: &str, raw: &str) -> String {
let detail = extract_transport_detail(raw);
let class = if is_embedded_provider_failure(&detail) {
classify_composio_error(tool, &detail)
} else if is_gateway_transport_shape(raw) {
ComposioErrorClass::Gateway
} else {
classify_composio_error(tool, raw)
};
let body = match class {
ComposioErrorClass::InsufficientScope => format_insufficient_scope_message(tool, &detail),
ComposioErrorClass::RateLimited => format_rate_limited_message(tool, &detail),
ComposioErrorClass::Gateway => format!(
"Temporary gateway error while calling `{tool}`: {}",
summarize_gateway(raw)
),
ComposioErrorClass::Validation => format!("Invalid arguments for `{tool}`: {detail}"),
ComposioErrorClass::UpstreamProvider => {
format!("`{tool}` failed at the connected provider: {detail}")
}
ComposioErrorClass::ComposioPlatform => {
format!("Composio connection issue for `{tool}`: {detail}")
}
ComposioErrorClass::Other => format!("`{tool}` failed: {detail}"),
};
prefix_class(class, &body)
}
fn prefix_class(class: ComposioErrorClass, body: &str) -> String {
format!("[composio:error:{}] {}", class.as_str(), body)
}
fn format_insufficient_scope_message(tool: &str, detail: &str) -> String {
let toolkit = tool
.split('_')
.next()
.unwrap_or("integration")
.to_ascii_lowercase();
format!(
"`{tool}` was rejected because the connected {toolkit} account is missing required \
permissions ({detail}). Reconnect the integration in Settings → Skills and grant the \
scopes requested during OAuth."
)
}
fn format_rate_limited_message(tool: &str, detail: &str) -> String {
format!(
"`{tool}` hit an upstream rate limit ({detail}). Wait a minute and retry, or reduce \
call frequency — this is not an OpenHuman gateway outage."
)
}
fn is_validation_shape(lower: &str) -> bool {
lower.contains("invalid arguments")
|| lower.contains("missing required")
|| lower.contains("must not be empty")
|| lower.contains("required field")
|| lower.contains("bad request")
|| lower.contains("invalid date")
|| lower.contains("rfc 3339")
|| lower.contains("timemax")
|| lower.contains("timemin")
}
fn is_insufficient_scope_shape(lower: &str) -> bool {
lower.contains("insufficient authentication scopes")
|| lower.contains("insufficient scope")
|| lower.contains("insufficient permissions")
|| (lower.contains("403") && lower.contains("scope"))
|| lower.contains("invalid oauth scope")
}
fn is_rate_limited_shape(lower: &str) -> bool {
lower.contains("rate limit")
|| lower.contains("rate_limit")
|| lower.contains("ratelimited")
|| lower.contains("too many requests")
|| lower.contains("429")
}
fn is_composio_platform_shape(lower: &str) -> bool {
lower.contains("connection error, try to authenticate")
|| lower.contains("not enabled")
|| lower.contains("not connected")
|| lower.contains("token revoked")
}
fn is_gateway_transport_shape(lower: &str) -> bool {
lower.contains("backend returned 502")
|| lower.contains("502 bad gateway")
|| lower.contains("backend returned 503")
|| lower.contains("backend returned 504")
|| lower.contains("(502 ")
|| lower.contains("(503 ")
|| lower.contains("(504 ")
}
fn is_embedded_provider_failure(lower: &str) -> bool {
is_validation_shape(lower)
|| is_insufficient_scope_shape(lower)
|| is_rate_limited_shape(lower)
|| is_composio_platform_shape(lower)
|| lower.contains("composio")
|| lower.contains("google")
|| lower.contains("slack")
|| lower.contains("notion")
|| lower.contains("gmail")
|| lower.contains("fetch_type")
|| lower.contains("timemax")
|| lower.contains("timemin")
}
fn extract_transport_detail(raw: &str) -> String {
raw.split_once(": ")
.map(|(_, tail)| tail.to_string())
.unwrap_or_else(|| raw.to_string())
}
fn summarize_gateway(raw: &str) -> String {
if let Some(idx) = raw.find("Backend returned ") {
let rest = &raw[idx..];
if let Some(colon) = rest.rfind(": ") {
return rest[colon + 2..].trim().to_string();
}
return rest.trim().to_string();
}
raw.trim().to_string()
}
#[cfg(test)]
#[path = "error_mapping_tests.rs"]
mod tests;
src/openhuman/composio/error_mapping_tests.rs
+44
View File
@@ -0,0 +1,44 @@
use super::{classify_composio_error, remap_transport_error, ComposioErrorClass};
#[test]
fn classifies_gmail_insufficient_scope() {
let msg = "HTTP 403: Request had insufficient authentication scopes.";
assert_eq!(
classify_composio_error("GMAIL_FETCH_EMAILS", msg),
ComposioErrorClass::InsufficientScope
);
}
#[test]
fn classifies_slack_rate_limit() {
let msg = "Slack API error: ratelimited";
assert_eq!(
classify_composio_error("SLACK_FETCH_CONVERSATION_HISTORY", msg),
ComposioErrorClass::RateLimited
);
}
#[test]
fn embedded_provider_failure_in_502_body_is_not_gateway() {
let raw = "Backend returned 502 Bad Gateway for POST https://api.example.com/agent-integrations/composio/execute: \
timeMax must be RFC 3339 timestamp";
let mapped = remap_transport_error("GOOGLECALENDAR_EVENTS_LIST", raw);
assert!(
mapped.contains("[composio:error:"),
"expected classified prefix, got: {mapped}"
);
assert!(
!mapped.contains("[composio:error:gateway]"),
"provider-shaped 502 body must not be labeled gateway: {mapped}"
);
}
#[test]
fn true_gateway_stays_gateway_class() {
let raw = "Backend returned 502 Bad Gateway for POST https://api.example.com/x: upstream down";
let mapped = remap_transport_error("GMAIL_SEND_EMAIL", raw);
assert!(
mapped.contains("[composio:error:gateway]"),
"expected gateway class, got: {mapped}"
);
}
src/openhuman/composio/execute_dispatch.rs
+188
View File
@@ -0,0 +1,188 @@
//! Shared Composio execute path: prepare args, retry policy, error mapping (#1797).
use std::time::Duration;
use super::auth_retry::{execute_with_auth_retry_inner, AUTH_RETRY_BACKOFF};
use super::client::{direct_execute, ComposioClient, ComposioClientKind};
use super::error_mapping::{format_provider_error, remap_transport_error};
use super::execute_prepare::prepare_execute_arguments;
use super::types::ComposioExecuteResponse;
const SLACK_HISTORY: &str = "SLACK_FETCH_CONVERSATION_HISTORY";
const RATELIMIT_INITIAL_BACKOFF: Duration = Duration::from_secs(2);
const RATELIMIT_MAX_BACKOFF: Duration = Duration::from_secs(30);
const RATELIMIT_MAX_ATTEMPTS: u32 = 6;
pub async fn execute_composio_action(
client: &ComposioClient,
tool: &str,
arguments: Option<serde_json::Value>,
) -> Result<ComposioExecuteResponse, String> {
let tool = tool.trim();
if tool.is_empty() {
return Err("composio: tool slug must not be empty".to_string());
}
let prepared = match prepare_execute_arguments(tool, arguments) {
Ok(args) => args,
Err(msg) => {
tracing::debug!(
tool = %tool,
error = %msg,
"[composio][prepare] local validation rejected execute"
);
return Err(format_provider_error(tool, &msg));
}
};
tracing::debug!(tool = %tool, "[composio][dispatch] execute_composio_action");
let resp = match execute_with_retries(client, tool, prepared).await {
Ok(resp) => resp,
Err(e) => {
tracing::debug!(tool = %tool, "[composio][dispatch] transport failure");
return Err(remap_transport_error(tool, &e.to_string()));
}
};
if resp.successful {
return Ok(resp);
}
let raw_err = resp
.error
.clone()
.unwrap_or_else(|| "provider reported failure".to_string());
Ok(ComposioExecuteResponse {
error: Some(format_provider_error(tool, &raw_err)),
..resp
})
}
async fn execute_with_retries(
client: &ComposioClient,
tool: &str,
args: serde_json::Value,
) -> anyhow::Result<ComposioExecuteResponse> {
let mut delay = RATELIMIT_INITIAL_BACKOFF;
for attempt in 1..=RATELIMIT_MAX_ATTEMPTS {
let resp = execute_with_auth_retry_inner(
client,
tool,
Some(args.clone()),
if attempt == 1 {
AUTH_RETRY_BACKOFF
} else {
Duration::ZERO
},
)
.await?;
if resp.successful {
return Ok(resp);
}
let err_text = resp.error.as_deref().unwrap_or("");
// Only Slack's conversations.history is allow-listed for transparent
// rate-limit retries today: it surfaces 429s on bursty agent reads and
// has stable retry semantics. Other tools surface 429 to the caller
// (formatted as `[composio:error:rate_limited]`) instead of stalling.
if tool == SLACK_HISTORY && is_rate_limited(err_text) && attempt < RATELIMIT_MAX_ATTEMPTS {
tracing::warn!(
tool = %tool,
attempt,
max_attempts = RATELIMIT_MAX_ATTEMPTS,
sleep_ms = delay.as_millis() as u64,
"[composio][dispatch] upstream rate limit; backing off (#1797)"
);
tokio::time::sleep(delay).await;
delay = (delay * 2).min(RATELIMIT_MAX_BACKOFF);
continue;
}
return Ok(resp);
}
unreachable!("loop returns on final attempt");
}
fn is_rate_limited(err: &str) -> bool {
let lower = err.to_ascii_lowercase();
lower.contains("rate limit")
|| lower.contains("rate_limit")
|| lower.contains("ratelimited")
|| lower.contains("too many requests")
|| lower.contains("429")
}
/// Mode-aware variant: routes through the backend (with auth-retry +
/// rate-limit backoff + error mapping) or the direct tenant client
/// (no auth-retry, but local validation + error mapping still apply).
///
/// Added after #1710's mode-aware client split (`ComposioClientKind`) so
/// the per-action tool surface, dispatcher tool, and RPC `composio_execute`
/// op all share one entry point with consistent error semantics.
pub async fn execute_composio_action_kind(
kind: ComposioClientKind,
tool: &str,
arguments: Option<serde_json::Value>,
entity_id: &str,
) -> Result<ComposioExecuteResponse, String> {
let tool_trim = tool.trim();
if tool_trim.is_empty() {
return Err("composio: tool slug must not be empty".to_string());
}
let prepared = match prepare_execute_arguments(tool_trim, arguments) {
Ok(args) => args,
Err(msg) => {
tracing::debug!(
tool = %tool_trim,
error = %msg,
"[composio][prepare] local validation rejected execute"
);
return Err(format_provider_error(tool_trim, &msg));
}
};
match kind {
ComposioClientKind::Backend(client) => {
tracing::debug!(tool = %tool_trim, "[composio][dispatch] backend variant");
let resp = match execute_with_retries(&client, tool_trim, prepared).await {
Ok(resp) => resp,
Err(e) => {
tracing::debug!(tool = %tool_trim, "[composio][dispatch] transport failure");
return Err(remap_transport_error(tool_trim, &e.to_string()));
}
};
Ok(format_response(tool_trim, resp))
}
ComposioClientKind::Direct(direct) => {
tracing::debug!(tool = %tool_trim, "[composio][dispatch] direct variant");
// Direct path skips auth_retry — the user's stored Composio
// API key has no backend-side refresh surface and a 401 is a
// config issue that should surface immediately rather than
// retry-loop. Local validation + error mapping still apply.
match direct_execute(&direct, tool_trim, Some(prepared), entity_id).await {
Ok(resp) => Ok(format_response(tool_trim, resp)),
Err(e) => Err(remap_transport_error(tool_trim, &e.to_string())),
}
}
}
}
fn format_response(tool: &str, resp: ComposioExecuteResponse) -> ComposioExecuteResponse {
if resp.successful {
return resp;
}
let raw_err = resp
.error
.clone()
.unwrap_or_else(|| "provider reported failure".to_string());
ComposioExecuteResponse {
error: Some(format_provider_error(tool, &raw_err)),
..resp
}
}
#[cfg(test)]
#[path = "execute_dispatch_tests.rs"]
mod tests;
src/openhuman/composio/execute_dispatch_tests.rs
+56
View File
@@ -0,0 +1,56 @@
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use axum::{routing::post, Json, Router};
use serde_json::json;
use super::execute_composio_action;
use crate::openhuman::composio::client::ComposioClient;
use crate::openhuman::integrations::IntegrationClient;
async fn start_mock_backend(app: Router) -> String {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
// The spawned task is intentionally orphaned; it dies with the tokio
// runtime when the test finishes. The previous oneshot-based graceful
// shutdown was broken because the sender was dropped immediately,
// signalling shutdown before the test could exercise the server.
tokio::spawn(async move {
let _ = axum::serve(listener, app).await;
});
format!("http://{addr}")
}
fn build_client(base: &str) -> ComposioClient {
let inner = Arc::new(IntegrationClient::new(
base.to_string(),
"test-token".to_string(),
));
ComposioClient::new(inner)
}
#[tokio::test]
async fn local_validation_skips_network() {
let attempts = Arc::new(AtomicUsize::new(0));
let app = Router::new().route(
"/agent-integrations/composio/execute",
post({
let attempts = attempts.clone();
move || async move {
attempts.fetch_add(1, Ordering::SeqCst);
Json(json!({"success": true, "data": {"successful": true, "data": {}, "costUsd": 0.0}}))
}
}),
);
let base = start_mock_backend(app).await;
let client = build_client(&base);
let err = execute_composio_action(
&client,
"GMAIL_SEND_EMAIL",
Some(json!({ "subject": "hello" })),
)
.await
.unwrap_err();
assert!(err.contains("[composio:error:"));
assert_eq!(attempts.load(Ordering::SeqCst), 0);
}
src/openhuman/composio/execute_prepare.rs
+163
View File
@@ -0,0 +1,163 @@
//! Normalize and validate Composio action arguments before dispatch (#1797).
use serde_json::{json, Map, Value};
pub fn prepare_execute_arguments(tool: &str, arguments: Option<Value>) -> Result<Value, String> {
let tool = tool.trim();
let mut args = match arguments {
Some(Value::Object(map)) => Value::Object(map),
Some(Value::Null) | None => Value::Object(Map::new()),
Some(other) => {
return Err(format!(
"composio: `{tool}` arguments must be a JSON object, got {}",
other
));
}
};
if tool.starts_with("GOOGLECALENDAR_") {
normalize_calendar_time_bounds(&mut args)?;
}
if tool == "NOTION_FETCH_DATA" {
ensure_notion_fetch_type(&mut args)?;
}
if tool == "GMAIL_SEND_EMAIL" {
validate_gmail_send_email(&args)?;
}
if tool == "GMAIL_ADD_LABEL_TO_EMAIL" {
validate_gmail_add_label(&args)?;
}
Ok(args)
}
fn normalize_calendar_time_bounds(args: &mut Value) -> Result<(), String> {
let Some(obj) = args.as_object_mut() else {
return Ok(());
};
for key in ["timeMin", "timeMax", "time_min", "time_max"] {
if let Some(v) = obj.get(key).cloned() {
if let Some(normalized) = normalize_rfc3339_bound(&v) {
obj.insert(key.to_string(), Value::String(normalized));
} else if v.is_string() {
return Err(format!(
"GOOGLECALENDAR time bound `{key}` must be an RFC 3339 timestamp \
(e.g. 2026-05-14T00:00:00Z), not a bare date"
));
}
}
}
Ok(())
}
fn normalize_rfc3339_bound(value: &Value) -> Option<String> {
let s = value.as_str()?.trim();
if s.is_empty() {
return None;
}
if s.contains('T') {
return Some(s.to_string());
}
// A bare date like `2026-05-14` is promoted to RFC 3339 midnight UTC.
// Parse explicitly so impossible dates such as `2026-99-99` are rejected
// up front instead of being passed through to Google Calendar.
if chrono::NaiveDate::parse_from_str(s, "%Y-%m-%d").is_ok() {
return Some(format!("{s}T00:00:00Z"));
}
None
}
fn ensure_notion_fetch_type(args: &mut Value) -> Result<(), String> {
let Some(obj) = args.as_object_mut() else {
return Ok(());
};
let has_fetch_type = obj
.get("fetch_type")
.or_else(|| obj.get("fetchType"))
.and_then(|v| v.as_str())
.map(str::trim)
.is_some_and(|s| !s.is_empty());
if has_fetch_type {
return Ok(());
}
let inferred = obj
.get("filter")
.and_then(|f| f.get("value").or_else(|| f.get("property")))
.and_then(|v| v.as_str())
.map(str::trim)
.filter(|s| !s.is_empty())
.map(|v| match v {
"page" | "pages" => "pages",
"database" | "databases" => "databases",
other => other,
})
.unwrap_or("pages");
tracing::debug!(
fetch_type = %inferred,
"[composio][prepare] NOTION_FETCH_DATA: inferred fetch_type"
);
obj.insert("fetch_type".to_string(), json!(inferred));
Ok(())
}
fn validate_gmail_send_email(args: &Value) -> Result<(), String> {
let Some(obj) = args.as_object() else {
return Err("GMAIL_SEND_EMAIL: arguments must be an object".to_string());
};
let recipient = obj
.get("to")
.or_else(|| obj.get("recipient_email"))
.or_else(|| obj.get("recipientEmail"))
.and_then(|v| v.as_str())
.map(str::trim)
.filter(|s| !s.is_empty());
if recipient.is_some() {
return Ok(());
}
Err(
"GMAIL_SEND_EMAIL: `to` (or `recipient_email`) is required — cannot send without a recipient"
.to_string(),
)
}
fn validate_gmail_add_label(args: &Value) -> Result<(), String> {
let Some(obj) = args.as_object() else {
return Err("GMAIL_ADD_LABEL_TO_EMAIL: arguments must be an object".to_string());
};
let message_id = obj
.get("message_id")
.or_else(|| obj.get("messageId"))
.and_then(|v| v.as_str())
.map(str::trim)
.filter(|s| !s.is_empty());
if message_id.is_none() {
return Err("GMAIL_ADD_LABEL_TO_EMAIL: `message_id` is required".to_string());
}
let add = non_empty_string_array(obj.get("add_label_ids").or_else(|| obj.get("addLabelIds")));
let remove = non_empty_string_array(
obj.get("remove_label_ids")
.or_else(|| obj.get("removeLabelIds")),
);
if add || remove {
return Ok(());
}
Err(
"GMAIL_ADD_LABEL_TO_EMAIL: provide at least one non-empty label in `add_label_ids` or \
`remove_label_ids`"
.to_string(),
)
}
fn non_empty_string_array(value: Option<&Value>) -> bool {
match value {
Some(Value::Array(items)) => items
.iter()
.any(|v| v.as_str().map(str::trim).is_some_and(|s| !s.is_empty())),
Some(Value::String(s)) => !s.trim().is_empty(),
_ => false,
}
}
#[cfg(test)]
#[path = "execute_prepare_tests.rs"]
mod tests;
src/openhuman/composio/execute_prepare_tests.rs
+41
View File
@@ -0,0 +1,41 @@
use serde_json::json;
use super::prepare_execute_arguments;
#[test]
fn calendar_bare_date_becomes_rfc3339() {
let args = json!({
"timeMin": "2026-05-14",
"timeMax": "2026-05-15"
});
let prepared = prepare_execute_arguments("GOOGLECALENDAR_EVENTS_LIST", Some(args)).unwrap();
assert_eq!(prepared["timeMin"], "2026-05-14T00:00:00Z");
assert_eq!(prepared["timeMax"], "2026-05-15T00:00:00Z");
}
#[test]
fn notion_fetch_data_infers_fetch_type_from_filter() {
let args = json!({
"filter": { "value": "page", "property": "object" },
"page_size": 25
});
let prepared = prepare_execute_arguments("NOTION_FETCH_DATA", Some(args)).unwrap();
assert_eq!(prepared["fetch_type"], "pages");
}
#[test]
fn gmail_send_requires_recipient() {
let err = prepare_execute_arguments("GMAIL_SEND_EMAIL", Some(json!({ "subject": "hi" })))
.unwrap_err();
assert!(err.contains("recipient") || err.contains("`to`"));
}
#[test]
fn gmail_add_label_requires_label_ids() {
let err = prepare_execute_arguments(
"GMAIL_ADD_LABEL_TO_EMAIL",
Some(json!({ "message_id": "m1", "add_label_ids": [], "remove_label_ids": [] })),
)
.unwrap_err();
assert!(err.contains("add_label_ids") || err.contains("remove_label_ids"));
}
src/openhuman/composio/mod.rs
+3
View File
@@ -39,6 +39,9 @@ pub mod action_tool;
pub mod auth_retry;
pub mod bus;
pub mod client;
pub mod error_mapping;
pub mod execute_dispatch;
pub mod execute_prepare;
pub mod googlecalendar_args;
pub mod ops;
pub mod periodic;
src/openhuman/composio/ops.rs
+29 -13
View File
@@ -22,8 +22,8 @@ type OpResult<T> = std::result::Result<T, String>;
use std::sync::Arc;
use super::client::{
build_composio_client, create_composio_client, direct_authorize, direct_execute,
direct_list_connections, direct_list_tools, ComposioClient, ComposioClientKind,
build_composio_client, create_composio_client, direct_authorize, direct_list_connections,
direct_list_tools, ComposioClient, ComposioClientKind,
};
use super::providers::{
get_provider, ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason,
@@ -571,16 +571,18 @@ pub async fn composio_execute(
// body preference, and cost-USD log line all stay uniform (#1710).
let kind = create_composio_client(config).map_err(|e| format!("[composio] execute: {e}"))?;
let started = std::time::Instant::now();
let result = match kind {
ComposioClientKind::Backend(client) => {
tracing::debug!(tool = %tool, "[composio] execute: backend variant");
client.execute_tool(tool, arguments).await
}
ComposioClientKind::Direct(direct) => {
tracing::debug!(tool = %tool, "[composio-direct] execute: direct variant");
direct_execute(&direct, tool, arguments, &config.composio.entity_id).await
}
};
// Centralized prepare → retry → error-mapping pipeline (#1797),
// mode-aware over the backend/direct split (#1710). The dispatcher
// returns pre-formatted `[composio:error:<class>] …` strings so the
// frontend formatter at `app/src/lib/composio/formatters.ts` can
// parse the class regardless of which mode produced the failure.
let result = super::execute_dispatch::execute_composio_action_kind(
kind,
tool,
arguments,
&config.composio.entity_id,
)
.await;
let elapsed_ms = started.elapsed().as_millis() as u64;
match result {
@@ -616,7 +618,21 @@ pub async fn composio_execute(
},
);
report_composio_op_error("execute", &e);
Err(format!("[composio] execute failed: {e:#}"))
// Preserve already-classified errors from the dispatcher
// (`[composio:error:<class>] …`) so the frontend formatter at
// `app/src/lib/composio/formatters.ts` can still parse the class.
let is_classified = e.starts_with("[composio:error:");
tracing::debug!(
tool = %tool,
elapsed_ms,
classified = is_classified,
"[composio] rpc execute error mapped"
);
if is_classified {
Err(e)
} else {
Err(format!("[composio] execute failed: {e}"))
}
}
}
}
src/openhuman/composio/ops_tests.rs
+9 -1
View File
@@ -405,7 +405,15 @@ async fn composio_execute_via_mock_propagates_backend_error() {
let err = composio_execute(&config, "ANY_TOOL", None)
.await
.unwrap_err();
assert!(err.contains("execute failed"));
// The dispatcher (`execute_composio_action`) classifies transport
// failures and prefixes them with `[composio:error:<class>] …`; ops.rs
// preserves that prefix so the frontend formatter can parse the class.
// For an unrecognised tool slug and a 502-shaped envelope the only
// signal we get is the backend error text, so assert on its contents.
assert!(
err.starts_with("[composio:error:") && err.contains("rate limited"),
"got: {err}"
);
}
#[tokio::test]
src/openhuman/composio/tools.rs
+11 -22
View File
@@ -34,9 +34,7 @@ use crate::openhuman::tools::traits::{
PermissionLevel, Tool, ToolCallOptions, ToolCategory, ToolResult,
};
use super::client::{
create_composio_client, direct_execute, direct_list_connections, ComposioClientKind,
};
use super::client::{create_composio_client, direct_list_connections, ComposioClientKind};
use super::providers::{
catalog_for_toolkit, classify_unknown, find_curated, get_provider, load_user_scope_or_default,
toolkit_from_slug, ToolScope, UserScopePref,
@@ -952,24 +950,15 @@ impl Tool for ComposioExecuteTool {
};
let started = std::time::Instant::now();
let res = match kind {
ComposioClientKind::Backend(client) => {
tracing::debug!(tool = %tool, "[composio] tool execute.execute: backend variant");
// Backend path retains the upstream `auth_retry` wrapper
// so a 401 from a stale tinyhumans-tenant token is
// refreshed-and-replayed exactly once before surfacing.
super::auth_retry::execute_with_auth_retry(&client, &tool, arguments).await
}
ComposioClientKind::Direct(direct) => {
tracing::debug!(tool = %tool, "[composio] tool execute.execute: direct variant");
// Direct path skips `auth_retry`: the user's stored
// Composio API key doesn't have a backend-side refresh
// surface and a 401 here is a config issue (rotated key,
// wrong key, deleted) that should surface to the user
// rather than retry-loop.
direct_execute(&direct, &tool, arguments, &live_config.composio.entity_id).await
}
};
// Centralized prepare → retry → error-mapping pipeline (#1797),
// mode-aware over the backend/direct split (#1710).
let res = super::execute_dispatch::execute_composio_action_kind(
kind,
&tool,
arguments,
&live_config.composio.entity_id,
)
.await;
let elapsed_ms = started.elapsed().as_millis() as u64;
match res {
Ok(resp) => {
@@ -1012,7 +1001,7 @@ impl Tool for ComposioExecuteTool {
elapsed_ms,
},
);
Ok(ToolResult::error(format!("composio_execute failed: {e}")))
Ok(ToolResult::error(e))
}
}
}
src/openhuman/socket/event_handlers.rs
+17
View File
@@ -37,6 +37,23 @@ pub(super) fn handle_sio_event(
event_name,
payload.len()
);
// CodeRabbit #3250222027: even at debug level, raw bodies can leak
// PII / secrets / tokens. Log structural metadata (top-level shape +
// byte length) but never the raw text.
let payload_shape = match &data {
serde_json::Value::Object(map) => format!("object_keys={}", map.len()),
serde_json::Value::Array(arr) => format!("array_len={}", arr.len()),
serde_json::Value::String(_) => "string".to_string(),
serde_json::Value::Number(_) => "number".to_string(),
serde_json::Value::Bool(_) => "bool".to_string(),
serde_json::Value::Null => "null".to_string(),
};
log::debug!(
"[socket] event payload: name={} data_bytes={} shape={} preview_omitted=true",
event_name,
payload.len(),
payload_shape
);
log::debug!("[socket] event dispatch: name={}", event_name);
match event_name {