From 69afbe30777609b5dea8a753822a2e4b257e1132 Mon Sep 17 00:00:00 2001 From: CodeGhost21 <164498022+CodeGhost21@users.noreply.github.com> Date: Thu, 4 Jun 2026 11:01:50 +0530 Subject: [PATCH] fix(composio): expose daily-budget + sync-interval via env (#2437 F) (#3314) Co-authored-by: Steven Enamakel --- .env.example | 25 ++++ .../composio/providers/clickup/provider.rs | 7 +- .../composio/providers/github/provider.rs | 7 +- .../composio/providers/gmail/provider.rs | 6 +- .../composio/providers/linear/provider.rs | 7 +- .../memory_sync/composio/providers/mod.rs | 2 +- .../composio/providers/notion/provider.rs | 8 +- .../composio/providers/slack/provider.rs | 6 +- .../composio/providers/sync_state.rs | 109 +++++++++++++- .../memory_sync/composio/providers/traits.rs | 137 ++++++++++++++++++ 10 files changed, 293 insertions(+), 21 deletions(-) diff --git a/.env.example b/.env.example index 4cc90a6c9..db9f7b55f 100644 --- a/.env.example +++ b/.env.example @@ -299,6 +299,31 @@ OPENHUMAN_BUILD_SHA= # [optional] Default: true — set to false to disable anonymized analytics & crash reports OPENHUMAN_ANALYTICS_ENABLED=true +# --------------------------------------------------------------------------- +# Composio memory-sync (per-provider knobs) +# --------------------------------------------------------------------------- +# All values below are [optional] runtime overrides for memory-sync providers +# (gmail / slack / notion / clickup / github / linear). Compile-time defaults +# match what shipped before these vars existed; only set what you need to +# change. +# +# Daily request budget per (toolkit, connection) pair. Compile-time default is +# 500 — enough for steady-state but easy to exhaust mid-backfill on chatty +# workspaces. Raise to widen the cap; non-positive / non-numeric values fall +# back to 500 with a `warn`. +# OPENHUMAN_COMPOSIO_DAILY_REQUEST_LIMIT=2000 +# +# Per-toolkit periodic sync cadence (seconds). Compile-time defaults: gmail / +# slack = 900 (15m), notion / clickup / github / linear = 1800 (30m). Raise +# the interval to burn the daily budget more slowly. Non-positive / non- +# numeric values fall back to the compile-time default with a `warn`. +# OPENHUMAN_COMPOSIO_GMAIL_SYNC_INTERVAL_SECS=1800 +# OPENHUMAN_COMPOSIO_SLACK_SYNC_INTERVAL_SECS=3600 +# OPENHUMAN_COMPOSIO_NOTION_SYNC_INTERVAL_SECS=3600 +# OPENHUMAN_COMPOSIO_CLICKUP_SYNC_INTERVAL_SECS=3600 +# OPENHUMAN_COMPOSIO_GITHUB_SYNC_INTERVAL_SECS=3600 +# OPENHUMAN_COMPOSIO_LINEAR_SYNC_INTERVAL_SECS=3600 + # --------------------------------------------------------------------------- # Logging # --------------------------------------------------------------------------- diff --git a/src/openhuman/memory_sync/composio/providers/clickup/provider.rs b/src/openhuman/memory_sync/composio/providers/clickup/provider.rs index 548ed4bf2..0a8db0447 100644 --- a/src/openhuman/memory_sync/composio/providers/clickup/provider.rs +++ b/src/openhuman/memory_sync/composio/providers/clickup/provider.rs @@ -32,8 +32,9 @@ use super::{ingest::ingest_task_into_memory_tree, sync}; use crate::openhuman::config::Config; use crate::openhuman::memory_sync::composio::providers::sync_state::{extract_item_id, SyncState}; use crate::openhuman::memory_sync::composio::providers::{ - first_array_str, merge_extra, pick_str, ComposioProvider, CuratedTool, NormalizedTask, - ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason, TaskFetchFilter, TaskKind, + first_array_str, merge_extra, pick_str, resolve_sync_interval_secs, ComposioProvider, + CuratedTool, NormalizedTask, ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason, + TaskFetchFilter, TaskKind, }; pub(crate) const ACTION_GET_AUTHORIZED_USER: &str = "CLICKUP_GET_AUTHORIZED_USER"; @@ -91,7 +92,7 @@ impl ComposioProvider for ClickUpProvider { // 30 minutes — same cadence as Notion. ClickUp tasks change // more slowly than chat but faster than email, so this is in // the middle. - Some(30 * 60) + Some(resolve_sync_interval_secs("clickup", 30 * 60)) } async fn fetch_user_profile( diff --git a/src/openhuman/memory_sync/composio/providers/github/provider.rs b/src/openhuman/memory_sync/composio/providers/github/provider.rs index bd77902c7..db7996eb8 100644 --- a/src/openhuman/memory_sync/composio/providers/github/provider.rs +++ b/src/openhuman/memory_sync/composio/providers/github/provider.rs @@ -27,8 +27,9 @@ use super::ingest::ingest_issue_into_memory_tree; use super::sync; use crate::openhuman::memory_sync::composio::providers::sync_state::SyncState; use crate::openhuman::memory_sync::composio::providers::{ - merge_extra, pick_str, ComposioProvider, CuratedTool, GithubFetchMode, NormalizedTask, - ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason, TaskFetchFilter, TaskKind, + merge_extra, pick_str, resolve_sync_interval_secs, ComposioProvider, CuratedTool, + GithubFetchMode, NormalizedTask, ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason, + TaskFetchFilter, TaskKind, }; pub(crate) const ACTION_GET_AUTHENTICATED_USER: &str = "GITHUB_GET_THE_AUTHENTICATED_USER"; @@ -75,7 +76,7 @@ impl ComposioProvider for GitHubProvider { // 30 minutes — GitHub issues change less frequently than Slack // messages, so a half-hour cadence keeps the memory fresh without // hammering the search API. - Some(30 * 60) + Some(resolve_sync_interval_secs("github", 30 * 60)) } async fn fetch_user_profile( diff --git a/src/openhuman/memory_sync/composio/providers/gmail/provider.rs b/src/openhuman/memory_sync/composio/providers/gmail/provider.rs index 67f353d34..cad580abe 100644 --- a/src/openhuman/memory_sync/composio/providers/gmail/provider.rs +++ b/src/openhuman/memory_sync/composio/providers/gmail/provider.rs @@ -29,8 +29,8 @@ use super::ingest::ingest_page_into_memory_tree; use super::sync; use crate::openhuman::memory_sync::composio::providers::sync_state::{extract_item_id, SyncState}; use crate::openhuman::memory_sync::composio::providers::{ - pick_str, ComposioProvider, CuratedTool, ProviderContext, ProviderUserProfile, SyncOutcome, - SyncReason, + pick_str, resolve_sync_interval_secs, ComposioProvider, CuratedTool, ProviderContext, + ProviderUserProfile, SyncOutcome, SyncReason, }; const ACTION_GET_PROFILE: &str = "GMAIL_GET_PROFILE"; @@ -117,7 +117,7 @@ impl ComposioProvider for GmailProvider { } fn sync_interval_secs(&self) -> Option { - Some(15 * 60) + Some(resolve_sync_interval_secs("gmail", 15 * 60)) } fn post_process_action_result( diff --git a/src/openhuman/memory_sync/composio/providers/linear/provider.rs b/src/openhuman/memory_sync/composio/providers/linear/provider.rs index 292c12838..eaa5ab478 100644 --- a/src/openhuman/memory_sync/composio/providers/linear/provider.rs +++ b/src/openhuman/memory_sync/composio/providers/linear/provider.rs @@ -26,8 +26,9 @@ use super::{ingest::ingest_issue_into_memory_tree, sync}; use crate::openhuman::config::Config; use crate::openhuman::memory_sync::composio::providers::sync_state::{extract_item_id, SyncState}; use crate::openhuman::memory_sync::composio::providers::{ - merge_extra, pick_str, ComposioProvider, CuratedTool, NormalizedTask, ProviderContext, - ProviderUserProfile, SyncOutcome, SyncReason, TaskFetchFilter, TaskKind, + merge_extra, pick_str, resolve_sync_interval_secs, ComposioProvider, CuratedTool, + NormalizedTask, ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason, TaskFetchFilter, + TaskKind, }; const ACTION_LIST_USERS: &str = "LINEAR_LIST_LINEAR_USERS"; @@ -78,7 +79,7 @@ impl ComposioProvider for LinearProvider { fn sync_interval_secs(&self) -> Option { // 30 minutes — same cadence as ClickUp/Notion. Linear issues change // more slowly than chat but faster than email. - Some(30 * 60) + Some(resolve_sync_interval_secs("linear", 30 * 60)) } async fn fetch_user_profile( diff --git a/src/openhuman/memory_sync/composio/providers/mod.rs b/src/openhuman/memory_sync/composio/providers/mod.rs index 316b707eb..640b014a2 100644 --- a/src/openhuman/memory_sync/composio/providers/mod.rs +++ b/src/openhuman/memory_sync/composio/providers/mod.rs @@ -282,7 +282,7 @@ pub use registry::{ }; pub use scope_lookup::{curated_scope_for, toolkit_has_scope}; pub use tool_scope::{classify_unknown, find_curated, toolkit_from_slug, CuratedTool, ToolScope}; -pub use traits::ComposioProvider; +pub use traits::{resolve_sync_interval_secs, sync_interval_env_var, ComposioProvider}; pub use types::{ ComposioUsage, ComposioUsageHandle, GithubFetchMode, NormalizedTask, ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason, TaskContainer, TaskFetchFilter, TaskKind, diff --git a/src/openhuman/memory_sync/composio/providers/notion/provider.rs b/src/openhuman/memory_sync/composio/providers/notion/provider.rs index d44832166..ee7e4ea2d 100644 --- a/src/openhuman/memory_sync/composio/providers/notion/provider.rs +++ b/src/openhuman/memory_sync/composio/providers/notion/provider.rs @@ -23,9 +23,9 @@ use super::sync; use crate::openhuman::config::Config; use crate::openhuman::memory_sync::composio::providers::sync_state::{extract_item_id, SyncState}; use crate::openhuman::memory_sync::composio::providers::{ - first_array_str, merge_extra, pick_str, ComposioProvider, CuratedTool, NormalizedTask, - ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason, TaskContainer, TaskFetchFilter, - TaskKind, + first_array_str, merge_extra, pick_str, resolve_sync_interval_secs, ComposioProvider, + CuratedTool, NormalizedTask, ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason, + TaskContainer, TaskFetchFilter, TaskKind, }; use futures::StreamExt; @@ -83,7 +83,7 @@ impl ComposioProvider for NotionProvider { } fn sync_interval_secs(&self) -> Option { - Some(30 * 60) + Some(resolve_sync_interval_secs("notion", 30 * 60)) } async fn fetch_user_profile( diff --git a/src/openhuman/memory_sync/composio/providers/slack/provider.rs b/src/openhuman/memory_sync/composio/providers/slack/provider.rs index f2f0d7784..7ca8d4403 100644 --- a/src/openhuman/memory_sync/composio/providers/slack/provider.rs +++ b/src/openhuman/memory_sync/composio/providers/slack/provider.rs @@ -47,8 +47,8 @@ use super::users::SlackUsers; use crate::openhuman::composio::types::ComposioExecuteResponse; use crate::openhuman::memory_sync::composio::providers::sync_state::SyncState; use crate::openhuman::memory_sync::composio::providers::{ - pick_str, ComposioProvider, CuratedTool, ProviderContext, ProviderUserProfile, SyncOutcome, - SyncReason, + pick_str, resolve_sync_interval_secs, ComposioProvider, CuratedTool, ProviderContext, + ProviderUserProfile, SyncOutcome, SyncReason, }; /// Composio action slug for channel listing. @@ -252,7 +252,7 @@ impl ComposioProvider for SlackProvider { } fn sync_interval_secs(&self) -> Option { - Some(SYNC_INTERVAL_SECS) + Some(resolve_sync_interval_secs("slack", SYNC_INTERVAL_SECS)) } fn post_process_action_result( diff --git a/src/openhuman/memory_sync/composio/providers/sync_state.rs b/src/openhuman/memory_sync/composio/providers/sync_state.rs index 70d414ac9..da7c01cf3 100644 --- a/src/openhuman/memory_sync/composio/providers/sync_state.rs +++ b/src/openhuman/memory_sync/composio/providers/sync_state.rs @@ -29,8 +29,46 @@ use crate::openhuman::memory_store::MemoryClientRef; /// day. This covers the initial backfill case where there are thousands of /// unsynced items — after this many requests the provider yields and /// continues on the next day. +/// +/// Compile-time default. The runtime value used by [`DailyBudget::default`] +/// is resolved by [`resolved_daily_request_limit`], which honors the +/// `OPENHUMAN_COMPOSIO_DAILY_REQUEST_LIMIT` env var so operators can widen +/// (or tighten) the cap without recompiling. See the project `.env.example` +/// for documentation. pub const DEFAULT_DAILY_REQUEST_LIMIT: u32 = 500; +/// Environment variable read by [`resolved_daily_request_limit`] to override +/// [`DEFAULT_DAILY_REQUEST_LIMIT`]. Must parse as a positive `u32`; values +/// `< 1` or non-numeric content fall back to the default with a `warn`. +pub const ENV_DAILY_REQUEST_LIMIT: &str = "OPENHUMAN_COMPOSIO_DAILY_REQUEST_LIMIT"; + +/// Resolve the effective per-day request limit. Reads +/// [`ENV_DAILY_REQUEST_LIMIT`] if set; otherwise returns +/// [`DEFAULT_DAILY_REQUEST_LIMIT`]. A non-positive or unparseable value +/// is rejected with a `warn` log and the default is used — we never +/// silently honor `0` because that would freeze every provider's sync +/// from the first tick. +pub fn resolved_daily_request_limit() -> u32 { + match std::env::var(ENV_DAILY_REQUEST_LIMIT) { + Ok(s) => match s.trim().parse::() { + Ok(n) if n >= 1 => n, + _ => { + static WARNED: std::sync::Once = std::sync::Once::new(); + WARNED.call_once(|| { + tracing::warn!( + env = ENV_DAILY_REQUEST_LIMIT, + value = %s, + default = DEFAULT_DAILY_REQUEST_LIMIT, + "[composio:sync-state] env override not a positive u32; using default" + ); + }); + DEFAULT_DAILY_REQUEST_LIMIT + } + }, + Err(_) => DEFAULT_DAILY_REQUEST_LIMIT, + } +} + /// KV namespace under which all sync state keys live. Separate from the /// memory document namespaces (`skill-gmail`, etc.) to avoid collisions. pub const KV_NAMESPACE: &str = "composio-sync-state"; @@ -103,7 +141,7 @@ impl Default for DailyBudget { Self { date: today_str(), requests_used: 0, - limit: DEFAULT_DAILY_REQUEST_LIMIT, + limit: resolved_daily_request_limit(), } } } @@ -493,4 +531,73 @@ mod tests { assert_eq!(s1.kv_key(), s2.kv_key()); assert_eq!(s1.kv_key(), "gmail:conn_x"); } + + /// RAII guard that save→set→restore an env var so the test does not + /// leak state to sibling tests in the same process. + struct EnvGuard { + key: &'static str, + previous: Option, + } + + impl EnvGuard { + fn set(key: &'static str, value: &str) -> Self { + let previous = std::env::var(key).ok(); + std::env::set_var(key, value); + Self { key, previous } + } + fn unset(key: &'static str) -> Self { + let previous = std::env::var(key).ok(); + std::env::remove_var(key); + Self { key, previous } + } + } + + impl Drop for EnvGuard { + fn drop(&mut self) { + match self.previous.take() { + Some(v) => std::env::set_var(self.key, v), + None => std::env::remove_var(self.key), + } + } + } + + // Env-var override scenarios are bundled into one `#[test]` so they + // run sequentially within a single thread — `cargo test` parallelism + // across `#[test]` fns would race on `OPENHUMAN_COMPOSIO_DAILY_REQUEST_LIMIT`. + #[test] + fn resolved_daily_request_limit_honors_env() { + let _lock = crate::openhuman::config::TEST_ENV_LOCK.lock().unwrap(); + + // Unset → default. + let _g = EnvGuard::unset(ENV_DAILY_REQUEST_LIMIT); + assert_eq!(resolved_daily_request_limit(), DEFAULT_DAILY_REQUEST_LIMIT); + assert_eq!(DailyBudget::default().limit, DEFAULT_DAILY_REQUEST_LIMIT); + drop(_g); + + // Valid override widens the cap. + let _g = EnvGuard::set(ENV_DAILY_REQUEST_LIMIT, "5000"); + assert_eq!(resolved_daily_request_limit(), 5000); + assert_eq!(DailyBudget::default().limit, 5000); + drop(_g); + + // Trims surrounding whitespace. + let _g = EnvGuard::set(ENV_DAILY_REQUEST_LIMIT, " 750 "); + assert_eq!(resolved_daily_request_limit(), 750); + drop(_g); + + // Zero rejected — would otherwise freeze every sync. + let _g = EnvGuard::set(ENV_DAILY_REQUEST_LIMIT, "0"); + assert_eq!(resolved_daily_request_limit(), DEFAULT_DAILY_REQUEST_LIMIT); + drop(_g); + + // Non-numeric rejected. + let _g = EnvGuard::set(ENV_DAILY_REQUEST_LIMIT, "lots"); + assert_eq!(resolved_daily_request_limit(), DEFAULT_DAILY_REQUEST_LIMIT); + drop(_g); + + // Negative rejected (won't parse as u32). + let _g = EnvGuard::set(ENV_DAILY_REQUEST_LIMIT, "-1"); + assert_eq!(resolved_daily_request_limit(), DEFAULT_DAILY_REQUEST_LIMIT); + drop(_g); + } } diff --git a/src/openhuman/memory_sync/composio/providers/traits.rs b/src/openhuman/memory_sync/composio/providers/traits.rs index 9fd51d001..1a0dfcf2f 100644 --- a/src/openhuman/memory_sync/composio/providers/traits.rs +++ b/src/openhuman/memory_sync/composio/providers/traits.rs @@ -234,3 +234,140 @@ pub trait ComposioProvider: Send + Sync { Ok(()) } } + +/// Build the env var name read by [`resolve_sync_interval_secs`] for a +/// given toolkit slug. Exposed so tests (and `.env.example`) can stay in +/// lockstep with the runtime lookup without re-implementing the casing. +pub fn sync_interval_env_var(toolkit: &str) -> String { + format!( + "OPENHUMAN_COMPOSIO_{}_SYNC_INTERVAL_SECS", + toolkit.to_ascii_uppercase() + ) +} + +/// Resolve the effective periodic sync interval (seconds) for a provider. +/// Reads `OPENHUMAN_COMPOSIO__SYNC_INTERVAL_SECS` if set; +/// otherwise returns `default_secs`. A non-positive or unparseable value +/// is rejected with a `warn` and the default is used — `0` would burn the +/// scheduler in a tight loop, so it is never honoured. +/// +/// Each provider's `sync_interval_secs()` impl calls this with its own +/// compile-time default so operators can independently slow down a +/// chatty toolkit (e.g. Slack) without rebuilding. +pub fn resolve_sync_interval_secs(toolkit: &str, default_secs: u64) -> u64 { + let key = sync_interval_env_var(toolkit); + match std::env::var(&key) { + Ok(s) => match s.trim().parse::() { + Ok(n) if n >= 1 => n, + _ => { + static WARNED: std::sync::Once = std::sync::Once::new(); + WARNED.call_once(|| { + tracing::warn!( + env = %key, + value = %s, + default = default_secs, + "[composio:provider] sync-interval env override not a positive u64; using default" + ); + }); + default_secs + } + }, + Err(_) => default_secs, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn sync_interval_env_var_uppercases_slug() { + assert_eq!( + sync_interval_env_var("slack"), + "OPENHUMAN_COMPOSIO_SLACK_SYNC_INTERVAL_SECS" + ); + assert_eq!( + sync_interval_env_var("GitHub"), + "OPENHUMAN_COMPOSIO_GITHUB_SYNC_INTERVAL_SECS" + ); + } + + /// RAII guard for env var save/restore so the test does not leak + /// state to siblings within the same process. + struct EnvGuard { + key: String, + previous: Option, + } + + impl EnvGuard { + fn set(key: &str, value: &str) -> Self { + let previous = std::env::var(key).ok(); + std::env::set_var(key, value); + Self { + key: key.to_string(), + previous, + } + } + fn unset(key: &str) -> Self { + let previous = std::env::var(key).ok(); + std::env::remove_var(key); + Self { + key: key.to_string(), + previous, + } + } + } + + impl Drop for EnvGuard { + fn drop(&mut self) { + match self.previous.take() { + Some(v) => std::env::set_var(&self.key, v), + None => std::env::remove_var(&self.key), + } + } + } + + // Bundled into a single `#[test]` so cargo's per-test parallelism + // does not race on the shared env var. Each scenario explicitly + // drops its guard before the next so the env is in a known state. + #[test] + fn resolve_sync_interval_honors_per_toolkit_env() { + let _lock = crate::openhuman::config::TEST_ENV_LOCK.lock().unwrap(); + + let key = sync_interval_env_var("slack"); + let default = 15 * 60; + + // Unset → default. + let _g = EnvGuard::unset(&key); + assert_eq!(resolve_sync_interval_secs("slack", default), default); + drop(_g); + + // Valid override slows the cadence. + let _g = EnvGuard::set(&key, "3600"); + assert_eq!(resolve_sync_interval_secs("slack", default), 3600); + drop(_g); + + // Whitespace tolerated. + let _g = EnvGuard::set(&key, " 1800 "); + assert_eq!(resolve_sync_interval_secs("slack", default), 1800); + drop(_g); + + // Zero rejected (would spin the scheduler). + let _g = EnvGuard::set(&key, "0"); + assert_eq!(resolve_sync_interval_secs("slack", default), default); + drop(_g); + + // Garbage rejected. + let _g = EnvGuard::set(&key, "soon"); + assert_eq!(resolve_sync_interval_secs("slack", default), default); + drop(_g); + + // Per-toolkit scoping: a different toolkit's var does not bleed + // into slack's lookup. + let gmail_key = sync_interval_env_var("gmail"); + let _slack_unset = EnvGuard::unset(&key); + let _gmail_set = EnvGuard::set(&gmail_key, "120"); + assert_eq!(resolve_sync_interval_secs("slack", default), default); + assert_eq!(resolve_sync_interval_secs("gmail", default), 120); + } +}