fix(composio): expose daily-budget + sync-interval via env (#2437 F) (#3314)

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

.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
 # ---------------------------------------------------------------------------

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(

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(

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<u64> {
-        Some(15 * 60)
+        Some(resolve_sync_interval_secs("gmail", 15 * 60))
     }
 
     fn post_process_action_result(

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<u64> {
         // 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(

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,

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<u64> {
-        Some(30 * 60)
+        Some(resolve_sync_interval_secs("notion", 30 * 60))
     }
 
     async fn fetch_user_profile(

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<u64> {
-        Some(SYNC_INTERVAL_SECS)
+        Some(resolve_sync_interval_secs("slack", SYNC_INTERVAL_SECS))
     }
 
     fn post_process_action_result(

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::<u32>() {
+            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<String>,
+    }
+
+    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);
+    }
 }

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_<TOOLKIT>_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::<u64>() {
+            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<String>,
+    }
+
+    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);
+    }
+}