feat(memory-sources): default sync sources on with conservative caps, per-source settings & All In (#3304)

src/bin/slack_backfill.rs

@@ -451,6 +451,8 @@ async fn main() -> Result<()> {
                 toolkit: conn.toolkit.clone(),
                 connection_id: Some(conn.id.clone()),
                 usage: Default::default(),
+                max_items: None,
+                sync_depth_days: None,
             };
             match run_backfill_via_search(&ctx, cli.days).await {
                 Ok(outcome) => {
@@ -549,6 +551,8 @@ async fn main() -> Result<()> {
             toolkit: conn.toolkit.clone(),
             connection_id: Some(conn.id.clone()),
             usage: Default::default(),
+            max_items: None,
+            sync_depth_days: None,
         };
         match provider.sync(&ctx, SyncReason::Manual).await {
             Ok(outcome) => {

src/openhuman/about_app/catalog_data.rs

@@ -355,6 +355,25 @@ pub(super) const CAPABILITIES: &[Capability] = &[
         status: CapabilityStatus::Beta,
         privacy: GITHUB_REPO_SOURCE,
     },
+    Capability {
+        id: "intelligence.memory_source_sync_controls",
+        name: "Memory Source Sync Defaults & Controls",
+        domain: "memory_sources",
+        category: CapabilityCategory::Intelligence,
+        description: "Connected memory sources are enabled by default with conservative, \
+            per-kind sync caps so the first sync stays cheap (e.g. Gmail ~100 recent emails, \
+            GitHub repo 10 PRs / 10 issues / 50 commits, RSS 20 items). Each source row exposes \
+            an inline settings panel to adjust the limit fields that apply to its kind \
+            (max_items, sync_depth_days, max_prs/issues/commits, since_days). \
+            An \"All In\" action enables every source and removes the caps to build the richest \
+            memory graph, then triggers a full sync. Already-connected sources are migrated to \
+            the new defaults once.",
+        how_to: "Intelligence > Memory Sources — toggle a source, open its gear for per-source \
+            limits, or use \"All In\". Programmatic: openhuman.memory_sources_update and \
+            openhuman.memory_sources_apply_all_in (RPC).",
+        status: CapabilityStatus::Beta,
+        privacy: LOCAL_RAW,
+    },
     Capability {
         id: "intelligence.embedding_provider_config",
         name: "Configure Embedding Provider",

src/openhuman/about_app/catalog_tests.rs

@@ -106,6 +106,7 @@ fn catalog_includes_additional_user_facing_surfaces() {
         "intelligence.embedding_provider_config",
         "intelligence.embedding_provider_test",
         "intelligence.github_repo_memory_source",
+        "intelligence.memory_source_sync_controls",
         "conversation.subagent_mascots",
     ] {
         assert!(

src/openhuman/composio/ops.rs

@@ -1264,6 +1264,8 @@ pub async fn composio_get_user_profile(
         toolkit: toolkit.clone(),
         connection_id: Some(connection_id.to_string()),
         usage: Default::default(),
+        max_items: None,
+        sync_depth_days: None,
     };
 
     let profile = provider.fetch_user_profile(&ctx).await.map_err(|e| {
@@ -1336,6 +1338,8 @@ pub async fn composio_refresh_all_identities(
             toolkit: toolkit.clone(),
             connection_id: Some(connection_id.clone()),
             usage: Default::default(),
+            max_items: None,
+            sync_depth_days: None,
         };
 
         match provider.fetch_user_profile(&ctx).await {
@@ -1433,6 +1437,8 @@ pub async fn composio_sync(
         toolkit: toolkit.clone(),
         connection_id: Some(connection_id.to_string()),
         usage: Default::default(),
+        max_items: None,
+        sync_depth_days: None,
     };
     let started_at_ms = std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)

src/openhuman/config/schema/types.rs

@@ -419,6 +419,18 @@ pub struct Config {
 
     #[serde(default)]
     pub model_registry: Vec<ModelRegistryEntry>,
+
+    /// Migration version guard for `apply_composio_source_caps_migration`.
+    ///
+    /// The migration runs whenever this is `< CURRENT_CAPS_MIGRATION_VERSION`
+    /// (see `memory_sources::reconcile`), then is bumped to that version. Using a
+    /// monotonic version (rather than a bool) lets an improved migration re-run
+    /// once for installs that already ran an earlier revision. Defaults to `0`
+    /// (`#[serde(default)]`); the retired `composio_source_caps_migrated` bool is
+    /// silently ignored (Config does not `deny_unknown_fields`), so prior installs
+    /// re-run the current migration exactly once.
+    #[serde(default)]
+    pub composio_source_caps_migration_version: u32,
 }
 
 /// Shared default so `#[serde(default)]` and `Config::default()` stay in sync.
@@ -738,6 +750,7 @@ impl Default for Config {
             onboarding_completed: false,
             chat_onboarding_completed: false,
             model_registry: Vec::new(),
+            composio_source_caps_migration_version: 0,
         }
     }
 }

src/openhuman/memory_sources/mod.rs

@@ -25,10 +25,11 @@ pub mod sync;
 pub mod types;
 
 pub use registry::{
-    add_source, get_source, list_enabled_by_kind, list_sources,
-    remove_composio_source_by_connection_id, remove_source, update_source, upsert_composio_source,
-    MemorySourcePatch,
+    add_source, apply_all_in, get_source, list_enabled_by_kind, list_sources,
+    memory_sync_defaults_for_toolkit, remove_composio_source_by_connection_id, remove_source,
+    update_source, upsert_composio_source, MemorySourcePatch,
 };
+pub use rpc::apply_kind_defaults;
 pub use schemas::{
     all_controller_schemas as all_memory_sources_controller_schemas,
     all_registered_controllers as all_memory_sources_registered_controllers,

src/openhuman/memory_sources/reconcile.rs

@@ -3,11 +3,20 @@
 //! Called once at boot to ensure all active Composio sync targets have
 //! a corresponding `MemorySourceEntry` in config. This catches connections
 //! created before the memory_sources domain existed.
+//!
+//! Also owns the retroactive caps migration
+//! (`apply_composio_source_caps_migration`) that gives any cap-less Composio
+//! source — enabled or disabled — conservative per-toolkit caps.
 
 use crate::openhuman::config::rpc as config_rpc;
 use crate::openhuman::memory_sources::registry;
+use crate::openhuman::memory_sources::types::{MemorySourceEntry, SourceKind};
 use crate::openhuman::memory_sync::composio;
 
+/// Current version of the caps migration. Bump when the migration logic changes
+/// so installs that ran an earlier revision re-run it exactly once.
+const CURRENT_CAPS_MIGRATION_VERSION: u32 = 1;
+
 pub async fn ensure_composio_sources() {
     tracing::debug!("[memory_sources:reconcile] starting composio reconciliation");
 
@@ -67,6 +76,100 @@ pub async fn ensure_composio_sources() {
             "[memory_sources:reconcile] composio reconciliation complete"
         );
     }
+
+    // Run the one-time caps migration after the reconcile loop so any
+    // sources upserted just above are also considered.
+    if let Err(e) = apply_composio_source_caps_migration().await {
+        tracing::warn!(
+            error = %e,
+            "[memory_sources:reconcile] caps migration failed (non-fatal, will retry next time)"
+        );
+    }
+}
+
+/// Apply conservative default caps in-place to every cap-less source.
+///
+/// For a Composio source with no `max_items`/`sync_depth_days`, writes the
+/// per-toolkit defaults and enables it (a no-op when already enabled) — an
+/// already-enabled, cap-less source would otherwise sync at the provider's large
+/// internal ceiling instead of the cheap default. For other kinds, fills any unset
+/// kind-specific caps via `apply_kind_defaults`. User-customised caps (non-None)
+/// are never overwritten. Returns the number of Composio entries that received
+/// defaults. Pure (no I/O) so it can be unit-tested directly.
+fn apply_caps_defaults_to_entries(sources: &mut [MemorySourceEntry]) -> u32 {
+    let mut applied = 0u32;
+    for source in sources.iter_mut() {
+        match source.kind {
+            SourceKind::Composio => {
+                // Apply to enabled AND disabled cap-less sources; skip entries the
+                // user has already customised (any non-None cap).
+                if source.max_items.is_none() && source.sync_depth_days.is_none() {
+                    let toolkit = source.toolkit.as_deref().unwrap_or("");
+                    let (max_items, sync_depth_days) =
+                        registry::memory_sync_defaults_for_toolkit(toolkit);
+                    tracing::debug!(
+                        id = %source.id,
+                        toolkit = %toolkit,
+                        was_enabled = source.enabled,
+                        max_items = ?max_items,
+                        sync_depth_days = ?sync_depth_days,
+                        "[memory_sources:reconcile] caps migration: applying conservative defaults"
+                    );
+                    source.enabled = true;
+                    source.max_items = max_items;
+                    source.sync_depth_days = sync_depth_days;
+                    applied += 1;
+                }
+            }
+            // Apply non-composio kind defaults for entries with all-None caps.
+            _ => {
+                // Use the rpc::apply_kind_defaults helper so the same
+                // conservative values are applied consistently.
+                crate::openhuman::memory_sources::rpc::apply_kind_defaults(source);
+            }
+        }
+    }
+    applied
+}
+
+/// Retroactive migration: give any cap-less Composio source — enabled or
+/// disabled — conservative per-toolkit caps so its first sync stays cheap.
+///
+/// Version-gated by `Config.composio_source_caps_migration_version`: runs once per
+/// `CURRENT_CAPS_MIGRATION_VERSION` bump (installs that ran an earlier revision
+/// re-run it exactly once). Entries the user has already customised (non-None caps)
+/// are left untouched.
+pub async fn apply_composio_source_caps_migration() -> Result<(), String> {
+    let mut config = config_rpc::load_config_with_timeout().await?;
+
+    if config.composio_source_caps_migration_version >= CURRENT_CAPS_MIGRATION_VERSION {
+        tracing::debug!(
+            version = config.composio_source_caps_migration_version,
+            "[memory_sources:reconcile] caps migration already at current version; skipping"
+        );
+        return Ok(());
+    }
+
+    tracing::info!(
+        from_version = config.composio_source_caps_migration_version,
+        to_version = CURRENT_CAPS_MIGRATION_VERSION,
+        "[memory_sources:reconcile] applying composio source caps migration"
+    );
+
+    let migrated_count = apply_caps_defaults_to_entries(&mut config.memory_sources);
+
+    config.composio_source_caps_migration_version = CURRENT_CAPS_MIGRATION_VERSION;
+    config
+        .save()
+        .await
+        .map_err(|e| format!("caps migration: failed to save config: {e:#}"))?;
+
+    tracing::info!(
+        migrated = migrated_count,
+        "[memory_sources:reconcile] caps migration complete"
+    );
+
+    Ok(())
 }
 
 fn title_case(s: &str) -> String {
@@ -92,6 +195,109 @@ fn short_id(id: &str) -> &str {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::openhuman::memory_sources::types::{MemorySourceEntry, SourceKind};
+
+    fn make_composio_entry(
+        id: &str,
+        toolkit: &str,
+        enabled: bool,
+        max_items: Option<u32>,
+        sync_depth_days: Option<u32>,
+    ) -> MemorySourceEntry {
+        MemorySourceEntry {
+            id: id.to_string(),
+            kind: SourceKind::Composio,
+            label: toolkit.to_string(),
+            enabled,
+            toolkit: Some(toolkit.to_string()),
+            connection_id: Some(format!("conn_{id}")),
+            path: None,
+            glob: None,
+            url: None,
+            branch: None,
+            paths: Vec::new(),
+            max_commits: None,
+            max_issues: None,
+            max_prs: None,
+            query: None,
+            since_days: None,
+            max_items,
+            selector: None,
+            max_tokens_per_sync: None,
+            max_cost_per_sync_usd: None,
+            sync_depth_days,
+        }
+    }
+
+    /// Exercises the real migration transform (`apply_caps_defaults_to_entries`)
+    /// so the tests cannot drift from the production predicate.
+    fn run_migration_on_entries(sources: &mut Vec<MemorySourceEntry>) -> u32 {
+        apply_caps_defaults_to_entries(sources)
+    }
+
+    #[test]
+    fn migration_flips_disabled_capless_entry_to_enabled_with_caps() {
+        let mut sources = vec![make_composio_entry("s1", "gmail", false, None, None)];
+        let count = run_migration_on_entries(&mut sources);
+        assert_eq!(count, 1);
+        assert!(sources[0].enabled);
+        assert_eq!(sources[0].max_items, Some(100));
+        assert_eq!(sources[0].sync_depth_days, Some(30));
+    }
+
+    #[test]
+    fn migration_applies_defaults_to_enabled_capless_entry() {
+        // An already-enabled but cap-less source must also receive defaults —
+        // otherwise its first sync runs at the provider's large internal ceiling.
+        let mut sources = vec![make_composio_entry("s2", "slack", true, None, None)];
+        let count = run_migration_on_entries(&mut sources);
+        assert_eq!(count, 1);
+        assert!(sources[0].enabled);
+        assert_eq!(sources[0].max_items, Some(50));
+        assert_eq!(sources[0].sync_depth_days, Some(14));
+    }
+
+    #[test]
+    fn migration_leaves_user_customised_caps_untouched() {
+        // User set max_items explicitly → migration should not override.
+        let mut sources = vec![make_composio_entry("s3", "notion", false, Some(5), None)];
+        let count = run_migration_on_entries(&mut sources);
+        assert_eq!(count, 0, "entry with user-set caps must not be migrated");
+        assert!(!sources[0].enabled, "enabled must not be flipped");
+        assert_eq!(sources[0].max_items, Some(5), "user cap must be preserved");
+    }
+
+    #[test]
+    fn migration_is_noop_on_empty_list() {
+        let mut sources: Vec<MemorySourceEntry> = vec![];
+        let count = run_migration_on_entries(&mut sources);
+        assert_eq!(count, 0);
+    }
+
+    #[test]
+    fn migration_applies_correct_defaults_per_toolkit() {
+        let toolkits = [
+            ("gmail", Some(100u32), Some(30u32)),
+            ("slack", Some(50), Some(14)),
+            ("notion", Some(30), Some(30)),
+            ("linear", Some(50), Some(30)),
+            ("clickup", Some(50), Some(30)),
+            ("github", Some(50), Some(30)),
+            ("unknown", Some(30), Some(14)),
+        ];
+        for (toolkit, exp_items, exp_days) in &toolkits {
+            let mut sources = vec![make_composio_entry("sid", toolkit, false, None, None)];
+            run_migration_on_entries(&mut sources);
+            assert_eq!(
+                sources[0].max_items, *exp_items,
+                "max_items mismatch for toolkit={toolkit}"
+            );
+            assert_eq!(
+                sources[0].sync_depth_days, *exp_days,
+                "sync_depth_days mismatch for toolkit={toolkit}"
+            );
+        }
+    }
 
     #[test]
     fn short_id_truncates_ascii() {

src/openhuman/memory_sources/registry.rs

@@ -7,6 +7,27 @@
 use crate::openhuman::config::rpc as config_rpc;
 use crate::openhuman::memory_sources::types::{MemorySourceEntry, SourceKind};
 
+/// Conservative default sync caps for a Composio toolkit, keyed by toolkit slug.
+///
+/// Single source of truth for the cheap out-of-the-box sync volume. Applied to a
+/// source entry when it is first registered (`upsert_composio_source`, insert-only)
+/// and by the one-time caps migration (`reconcile::apply_composio_source_caps_migration`)
+/// for cap-less entries. Never overwrites a user-customised cap.
+///
+/// Returns `(max_items, sync_depth_days)`.
+pub fn memory_sync_defaults_for_toolkit(toolkit: &str) -> (Option<u32>, Option<u32>) {
+    match toolkit {
+        "gmail" => (Some(100), Some(30)),
+        "slack" => (Some(50), Some(14)),
+        "notion" => (Some(30), Some(30)),
+        "linear" => (Some(50), Some(30)),
+        "clickup" => (Some(50), Some(30)),
+        "github" => (Some(50), Some(30)),
+        // Generic fallback for any toolkit not listed above.
+        _ => (Some(30), Some(14)),
+    }
+}
+
 pub async fn list_sources() -> Result<Vec<MemorySourceEntry>, String> {
     let config = config_rpc::load_config_with_timeout().await?;
     Ok(config.memory_sources.clone())
@@ -38,7 +59,6 @@ pub async fn add_source(entry: MemorySourceEntry) -> Result<MemorySourceEntry, S
     tracing::info!(
         id = %entry.id,
         kind = %entry.kind.as_str(),
-        label = %entry.label,
         "[memory_sources] adding source"
     );
 
@@ -111,6 +131,15 @@ pub async fn update_source(
     if let Some(v) = patch.sync_depth_days {
         entry.sync_depth_days = Some(v);
     }
+    if let Some(v) = patch.max_commits {
+        entry.max_commits = Some(v);
+    }
+    if let Some(v) = patch.max_issues {
+        entry.max_issues = Some(v);
+    }
+    if let Some(v) = patch.max_prs {
+        entry.max_prs = Some(v);
+    }
 
     entry.validate()?;
     let updated = entry.clone();
@@ -202,11 +231,19 @@ pub async fn upsert_composio_source(
         return Ok(updated);
     }
 
+    let (default_max_items, default_sync_depth_days) = memory_sync_defaults_for_toolkit(toolkit);
+    tracing::debug!(
+        toolkit = %toolkit,
+        max_items = ?default_max_items,
+        sync_depth_days = ?default_sync_depth_days,
+        "[memory_sources] applying conservative defaults for new composio source"
+    );
+
     let entry = MemorySourceEntry {
         id: format!("src_{}", uuid::Uuid::new_v4().as_simple()),
         kind: SourceKind::Composio,
         label: label.to_string(),
-        enabled: false,
+        enabled: true,
         toolkit: Some(toolkit.to_string()),
         connection_id: Some(connection_id.to_string()),
         path: None,
@@ -219,11 +256,11 @@ pub async fn upsert_composio_source(
         max_prs: None,
         query: None,
         since_days: None,
-        max_items: None,
+        max_items: default_max_items,
         selector: None,
         max_tokens_per_sync: None,
         max_cost_per_sync_usd: None,
-        sync_depth_days: None,
+        sync_depth_days: default_sync_depth_days,
     };
     config.memory_sources.push(entry.clone());
     config
@@ -275,12 +312,106 @@ pub struct MemorySourcePatch {
     pub max_cost_per_sync_usd: Option<f64>,
     #[serde(default)]
     pub sync_depth_days: Option<u32>,
+    // ── GithubRepo-specific caps (previously missing from patch) ──
+    #[serde(default)]
+    pub max_commits: Option<u32>,
+    #[serde(default)]
+    pub max_issues: Option<u32>,
+    #[serde(default)]
+    pub max_prs: Option<u32>,
+}
+
+/// Enable ALL configured memory sources and clear every per-source cap,
+/// giving the user unrestricted access ("All In" mode).
+///
+/// For each source in `config.memory_sources`:
+/// - Sets `enabled = true`.
+/// - Clears `max_items`, `since_days`, `sync_depth_days`,
+///   `max_commits`, `max_issues`, `max_prs`,
+///   `max_tokens_per_sync`, `max_cost_per_sync_usd` to `None`.
+///
+/// Saves config once after all mutations and returns the updated entries.
+pub async fn apply_all_in() -> Result<Vec<MemorySourceEntry>, String> {
+    let mut config = config_rpc::load_config_with_timeout().await?;
+
+    tracing::info!(
+        count = config.memory_sources.len(),
+        "[memory_sources] apply_all_in: enabling all sources and clearing caps"
+    );
+
+    for source in &mut config.memory_sources {
+        tracing::debug!(
+            id = %source.id,
+            kind = %source.kind.as_str(),
+            "[memory_sources] apply_all_in: enabling source and clearing caps"
+        );
+        source.enabled = true;
+        source.max_items = None;
+        source.since_days = None;
+        source.sync_depth_days = None;
+        source.max_commits = None;
+        source.max_issues = None;
+        source.max_prs = None;
+        source.max_tokens_per_sync = None;
+        source.max_cost_per_sync_usd = None;
+    }
+
+    let updated = config.memory_sources.clone();
+
+    config
+        .save()
+        .await
+        .map_err(|e| format!("apply_all_in: failed to save config: {e:#}"))?;
+
+    tracing::info!(
+        count = updated.len(),
+        "[memory_sources] apply_all_in: complete — all sources enabled, all caps cleared"
+    );
+
+    Ok(updated)
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    #[test]
+    fn composio_defaults_for_known_toolkits() {
+        assert_eq!(
+            memory_sync_defaults_for_toolkit("gmail"),
+            (Some(100), Some(30))
+        );
+        assert_eq!(
+            memory_sync_defaults_for_toolkit("slack"),
+            (Some(50), Some(14))
+        );
+        assert_eq!(
+            memory_sync_defaults_for_toolkit("notion"),
+            (Some(30), Some(30))
+        );
+        assert_eq!(
+            memory_sync_defaults_for_toolkit("linear"),
+            (Some(50), Some(30))
+        );
+        assert_eq!(
+            memory_sync_defaults_for_toolkit("clickup"),
+            (Some(50), Some(30))
+        );
+        assert_eq!(
+            memory_sync_defaults_for_toolkit("github"),
+            (Some(50), Some(30))
+        );
+    }
+
+    #[test]
+    fn composio_defaults_for_generic_fallback() {
+        assert_eq!(
+            memory_sync_defaults_for_toolkit("unknown_toolkit_xyz"),
+            (Some(30), Some(14))
+        );
+        assert_eq!(memory_sync_defaults_for_toolkit(""), (Some(30), Some(14)));
+    }
+
     #[test]
     fn memory_source_patch_deserializes_partial() {
         let json = serde_json::json!({ "label": "New label", "enabled": false });
@@ -289,4 +420,29 @@ mod tests {
         assert_eq!(patch.enabled, Some(false));
         assert!(patch.toolkit.is_none());
     }
+
+    #[test]
+    fn memory_source_patch_round_trips_github_limit_fields() {
+        let json = serde_json::json!({
+            "max_commits": 100,
+            "max_issues": 50,
+            "max_prs": 25
+        });
+        let patch: MemorySourcePatch = serde_json::from_value(json).unwrap();
+        assert_eq!(patch.max_commits, Some(100));
+        assert_eq!(patch.max_issues, Some(50));
+        assert_eq!(patch.max_prs, Some(25));
+        // Unset fields must be None (serde(default))
+        assert!(patch.label.is_none());
+        assert!(patch.enabled.is_none());
+    }
+
+    #[test]
+    fn memory_source_patch_defaults_github_fields_to_none() {
+        let json = serde_json::json!({ "enabled": true });
+        let patch: MemorySourcePatch = serde_json::from_value(json).unwrap();
+        assert!(patch.max_commits.is_none());
+        assert!(patch.max_issues.is_none());
+        assert!(patch.max_prs.is_none());
+    }
 }

src/openhuman/memory_sources/rpc.rs

@@ -105,7 +105,7 @@ pub async fn add_rpc(req: AddRequest) -> Result<RpcOutcome<AddResponse>, String>
         "[memory_sources] add_rpc: entry"
     );
 
-    let entry = MemorySourceEntry {
+    let mut entry = MemorySourceEntry {
         id: format!("src_{}", uuid::Uuid::new_v4().as_simple()),
         kind: req.kind,
         label: req.label,
@@ -117,9 +117,6 @@ pub async fn add_rpc(req: AddRequest) -> Result<RpcOutcome<AddResponse>, String>
         url: req.url,
         branch: req.branch,
         paths: req.paths,
-        // Per-type GitHub sync caps default to DEFAULT_GITHUB_ITEM_LIMIT
-        // (2000) in the reader when None. Overrides are applied via the
-        // update/patch path (`MemorySourcePatch`).
         max_commits: req.max_commits,
         max_issues: req.max_issues,
         max_prs: req.max_prs,
@@ -132,10 +129,48 @@ pub async fn add_rpc(req: AddRequest) -> Result<RpcOutcome<AddResponse>, String>
         sync_depth_days: req.sync_depth_days,
     };
 
+    // Apply conservative per-kind defaults when the caller left caps unset.
+    apply_kind_defaults(&mut entry);
+
     let source = registry::add_source(entry).await?;
     Ok(RpcOutcome::new(AddResponse { source }, vec![]))
 }
 
+/// Apply conservative per-kind cap defaults to a new source entry.
+///
+/// Only fills fields that are still `None` — never overwrites a
+/// caller-supplied value. This mirrors the retroactive migration logic in
+/// `reconcile::apply_composio_source_caps_migration` so the same defaults
+/// are applied consistently at creation time and during migration.
+pub fn apply_kind_defaults(entry: &mut MemorySourceEntry) {
+    match entry.kind {
+        SourceKind::GithubRepo => {
+            if entry.max_prs.is_none() {
+                entry.max_prs = Some(10);
+            }
+            if entry.max_issues.is_none() {
+                entry.max_issues = Some(10);
+            }
+            if entry.max_commits.is_none() {
+                entry.max_commits = Some(50);
+            }
+        }
+        SourceKind::RssFeed => {
+            if entry.max_items.is_none() {
+                entry.max_items = Some(20);
+            }
+        }
+        SourceKind::TwitterQuery => {
+            if entry.since_days.is_none() {
+                entry.since_days = Some(7);
+            }
+        }
+        // Folder / WebPage / Composio: no defaults to apply here.
+        // Composio defaults are set at upsert time in registry::upsert_composio_source.
+        _ => {}
+    }
+}
+
 // ── Update ──
 
 #[derive(Debug, serde::Deserialize)]
@@ -394,3 +429,72 @@ pub async fn monthly_cost_summary_rpc() -> Result<RpcOutcome<MonthlyCostSummaryR
         vec![],
     ))
 }
+
+// ── Apply All In ──
+
+/// Response returned by `memory_sources_apply_all_in`.
+#[derive(Debug, serde::Serialize)]
+pub struct AllInResponse {
+    /// All memory source entries after the "all in" transformation
+    /// (every source enabled, every cap cleared).
+    pub sources: Vec<MemorySourceEntry>,
+    /// Number of sync tasks spawned (one per enabled source).
+    pub sync_triggered: u32,
+}
+
+/// Enable ALL memory sources, clear all caps, and trigger a sync for
+/// every source.
+///
+/// Returns immediately with the updated source list and the number of
+/// syncs queued. Individual syncs run in the background and publish
+/// `MemorySyncStageChanged` events as they progress.
+pub async fn apply_all_in_rpc() -> Result<RpcOutcome<AllInResponse>, String> {
+    tracing::info!("[memory_sources] apply_all_in_rpc: entry");
+
+    // Enable all sources and clear caps.
+    let sources = registry::apply_all_in().await?;
+
+    // Trigger a background sync for every enabled source.
+    let config = config_rpc::load_config_with_timeout().await?;
+    let mut sync_triggered: u32 = 0;
+
+    for source in &sources {
+        if !source.enabled {
+            continue;
+        }
+        tracing::debug!(
+            source_id = %source.id,
+            kind = %source.kind.as_str(),
+            "[memory_sources] apply_all_in_rpc: triggering sync"
+        );
+        match crate::openhuman::memory_sources::sync::sync_source(source.clone(), config.clone())
+            .await
+        {
+            Ok(()) => {
+                sync_triggered += 1;
+            }
+            Err(e) => {
+                // Non-fatal: log and continue — best-effort sync trigger.
+                tracing::warn!(
+                    source_id = %source.id,
+                    error = %e,
+                    "[memory_sources] apply_all_in_rpc: sync trigger failed for source"
+                );
+            }
+        }
+    }
+
+    tracing::info!(
+        sources = sources.len(),
+        sync_triggered,
+        "[memory_sources] apply_all_in_rpc: complete"
+    );
+
+    Ok(RpcOutcome::new(
+        AllInResponse {
+            sources,
+            sync_triggered,
+        },
+        vec![],
+    ))
+}

src/openhuman/memory_sources/schemas.rs

@@ -55,6 +55,24 @@ fn kind_specific_fields() -> Vec<FieldSchema> {
             comment: "Path filters for github_repo sources.",
             required: false,
         },
+        FieldSchema {
+            name: "max_commits",
+            ty: TypeSchema::Option(Box::new(TypeSchema::U64)),
+            comment: "Max commits per sync for github_repo sources.",
+            required: false,
+        },
+        FieldSchema {
+            name: "max_issues",
+            ty: TypeSchema::Option(Box::new(TypeSchema::U64)),
+            comment: "Max issues per sync for github_repo sources.",
+            required: false,
+        },
+        FieldSchema {
+            name: "max_prs",
+            ty: TypeSchema::Option(Box::new(TypeSchema::U64)),
+            comment: "Max pull requests per sync for github_repo sources.",
+            required: false,
+        },
         FieldSchema {
             name: "query",
             ty: TypeSchema::Option(Box::new(TypeSchema::String)),
@@ -70,7 +88,7 @@ fn kind_specific_fields() -> Vec<FieldSchema> {
         FieldSchema {
             name: "max_items",
             ty: TypeSchema::Option(Box::new(TypeSchema::U64)),
-            comment: "Maximum items for rss_feed sources.",
+            comment: "Maximum items for rss_feed or composio sources.",
             required: false,
         },
         FieldSchema {
@@ -114,6 +132,7 @@ pub fn all_controller_schemas() -> Vec<ControllerSchema> {
         schemas("sync_audit_log"),
         schemas("estimate_sync_cost"),
         schemas("monthly_cost_summary"),
+        schemas("apply_all_in"),
     ]
 }
 
@@ -167,6 +186,10 @@ pub fn all_registered_controllers() -> Vec<RegisteredController> {
             schema: schemas("monthly_cost_summary"),
             handler: handle_monthly_cost_summary,
         },
+        RegisteredController {
+            schema: schemas("apply_all_in"),
+            handler: handle_apply_all_in,
+        },
     ]
 }
 
@@ -487,6 +510,29 @@ pub fn schemas(function: &str) -> ControllerSchema {
                 },
             ],
         },
+        "apply_all_in" => ControllerSchema {
+            namespace: NAMESPACE,
+            function: "apply_all_in",
+            description: "Enable ALL memory sources, clear all per-source caps, \
+                          and trigger a background sync for every source. \
+                          Returns immediately with the updated source list and \
+                          the count of sync tasks queued.",
+            inputs: vec![],
+            outputs: vec![
+                FieldSchema {
+                    name: "sources",
+                    ty: TypeSchema::Array(Box::new(TypeSchema::Ref("MemorySourceEntry"))),
+                    comment: "All memory sources after the all-in transformation.",
+                    required: true,
+                },
+                FieldSchema {
+                    name: "sync_triggered",
+                    ty: TypeSchema::U64,
+                    comment: "Number of sync tasks spawned.",
+                    required: true,
+                },
+            ],
+        },
         other => panic!("unknown memory_sources schema function: {other}"),
     }
 }
@@ -563,6 +609,10 @@ fn handle_monthly_cost_summary(_params: Map<String, Value>) -> ControllerFuture
     Box::pin(async move { to_json(rpc::monthly_cost_summary_rpc().await?) })
 }
 
+fn handle_apply_all_in(_params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move { to_json(rpc::apply_all_in_rpc().await?) })
+}
+
 fn parse_value<T: DeserializeOwned>(v: Value) -> Result<T, String> {
     serde_json::from_value(v).map_err(|e| format!("invalid params: {e}"))
 }

src/openhuman/memory_sync/composio/bus.rs

@@ -470,7 +470,33 @@ impl EventHandler for ComposioConnectionCreatedSubscriber {
                     return;
                 }
             };
-            let Some(ctx) = ProviderContext::from_config(
+            // Look up per-source caps from the memory_sources registry.
+            // Non-fatal: if the lookup fails we proceed without caps.
+            //
+            // upsert_composio_source runs AFTER this block (below), so for
+            // brand-new connections the entry may not exist yet. In that case
+            // fall back to the per-toolkit defaults so the first sync is still
+            // capped. list_enabled_by_kind would also drop disabled-but-
+            // configured entries, so we use list_sources() and filter ourselves.
+            let (src_max_items, src_sync_depth_days) = {
+                let registry_sources = crate::openhuman::memory_sources::list_sources()
+                    .await
+                    .unwrap_or_default();
+                registry_sources
+                    .iter()
+                    .find(|s| {
+                        s.kind == crate::openhuman::memory_sources::SourceKind::Composio
+                            && s.connection_id.as_deref() == Some(connection_id.as_str())
+                    })
+                    .map(|s| (s.max_items, s.sync_depth_days))
+                    .unwrap_or_else(|| {
+                        crate::openhuman::memory_sources::memory_sync_defaults_for_toolkit(
+                            toolkit.as_str(),
+                        )
+                    })
+            };
+
+            let Some(mut ctx) = ProviderContext::from_config(
                 Arc::new(config),
                 toolkit.clone(),
                 Some(connection_id.clone()),
@@ -482,6 +508,17 @@ impl EventHandler for ComposioConnectionCreatedSubscriber {
                 return;
             };
 
+            ctx.max_items = src_max_items;
+            ctx.sync_depth_days = src_sync_depth_days;
+
+            tracing::debug!(
+                toolkit = %toolkit,
+                connection_id = %connection_id,
+                max_items = ?src_max_items,
+                sync_depth_days = ?src_sync_depth_days,
+                "[composio:bus] caps from registry for connection_created"
+            );
+
             // `wait_for_connection_active` is a backend-only metadata
             // probe (`list_connections`). Resolve a backend
             // `ComposioClient` from the live config for it; direct-mode

src/openhuman/memory_sync/composio/mod.rs

@@ -151,11 +151,35 @@ pub async fn run_connection_sync(
         ))
     })?;
 
+    // Look up the source entry to obtain any user-configured caps.
+    // Non-fatal: if the registry read fails we proceed uncapped.
+    let (src_max_items, src_sync_depth_days) = {
+        let registry_sources = crate::openhuman::memory_sources::list_enabled_by_kind(
+            crate::openhuman::memory_sources::SourceKind::Composio,
+        )
+        .await
+        .unwrap_or_default();
+        registry_sources
+            .iter()
+            .find(|s| s.connection_id.as_deref() == Some(&target.connection_id))
+            .map(|s| (s.max_items, s.sync_depth_days))
+            .unwrap_or((None, None))
+    };
+
+    tracing::debug!(
+        connection_id = %target.connection_id,
+        max_items = ?src_max_items,
+        sync_depth_days = ?src_sync_depth_days,
+        "[composio:sync] run_connection_sync: caps from registry"
+    );
+
     let ctx = ProviderContext {
         config: std::sync::Arc::new(config),
         toolkit: target.toolkit,
         connection_id: Some(target.connection_id),
         usage: Default::default(),
+        max_items: src_max_items,
+        sync_depth_days: src_sync_depth_days,
     };
 
     let sync_result = provider.sync(&ctx, reason).await;

src/openhuman/memory_sync/composio/periodic.rs

@@ -313,6 +313,29 @@ pub(crate) async fn run_one_tick() -> Result<(), String> {
             continue;
         }
 
+        // Look up per-source caps from the memory_sources registry.
+        // Non-fatal: if the lookup fails we proceed without caps.
+        let (src_max_items, src_sync_depth_days) = {
+            let registry_sources = crate::openhuman::memory_sources::list_enabled_by_kind(
+                crate::openhuman::memory_sources::SourceKind::Composio,
+            )
+            .await
+            .unwrap_or_default();
+            registry_sources
+                .iter()
+                .find(|s| s.connection_id.as_deref() == Some(conn.id.as_str()))
+                .map(|s| (s.max_items, s.sync_depth_days))
+                .unwrap_or((None, None))
+        };
+
+        tracing::debug!(
+            toolkit = %toolkit,
+            connection_id = %conn.id,
+            max_items = ?src_max_items,
+            sync_depth_days = ?src_sync_depth_days,
+            "[composio:periodic] caps from registry"
+        );
+
         // Build a context tied to this specific connection and dispatch.
         // `ProviderContext` no longer caches a pre-baked
         // `ComposioClient` — provider methods resolve a fresh handle per
@@ -323,6 +346,8 @@ pub(crate) async fn run_one_tick() -> Result<(), String> {
             toolkit: toolkit.clone(),
             connection_id: Some(conn.id.clone()),
             usage: Default::default(),
+            max_items: src_max_items,
+            sync_depth_days: src_sync_depth_days,
         };
 
         tracing::debug!(

src/openhuman/memory_sync/composio/providers/clickup/provider.rs

@@ -261,12 +261,42 @@ impl ComposioProvider for ClickUpProvider {
             _ => PAGE_SIZE,
         };
 
+        // ctx.max_items: route through ItemCap — page ceiling, mid-page
+        // per-item break, and post-page hard stop all share one source of truth.
+        let mut cap = super::super::helpers::ItemCap::new(ctx.max_items);
+        let effective_max_pages = cap.max_pages(page_size, MAX_PAGES_PER_WORKSPACE);
+        if ctx.max_items.is_some() && effective_max_pages < MAX_PAGES_PER_WORKSPACE {
+            tracing::debug!(
+                connection_id = %connection_id,
+                max_items = ?ctx.max_items,
+                effective_max_pages,
+                "[composio:clickup] [memory_sync] applying max_items page cap"
+            );
+        }
+
+        // ctx.sync_depth_days: oldest allowed date_updated for client-side skip.
+        // ClickUp's `date_updated` field is a millisecond-epoch string, so the
+        // floor must also be epoch-millis (not RFC3339) for the lexicographic
+        // compare in `select_pending` to work correctly.
+        let oldest_allowed_time: Option<String> = ctx.sync_depth_days.map(|days| {
+            let floor = chrono::Utc::now() - chrono::Duration::days(days as i64);
+            let s = floor.timestamp_millis().to_string();
+            tracing::debug!(
+                connection_id = %connection_id,
+                sync_depth_days = days,
+                oldest_allowed_ms = %s,
+                "[composio:clickup] [memory_sync] applying sync_depth_days floor"
+            );
+            s
+        });
+
         let mut total_fetched: usize = 0;
         let mut total_persisted: usize = 0;
         let mut newest_updated: Option<String> = None;
+        let mut hit_cap_boundary = false;
 
         'workspaces: for workspace_id in &workspaces {
-            for page_num in 0..MAX_PAGES_PER_WORKSPACE {
+            for page_num in 0..effective_max_pages {
                 if state.budget_exhausted() {
                     tracing::info!(
                         workspace_id = %workspace_id,
@@ -327,9 +357,27 @@ impl ComposioProvider for ClickUpProvider {
                 }
 
                 // ── Per-item dedup + bounded-concurrency ingest ─────
-                let (pending, hit_cursor_boundary) =
+                let (mut pending, mut hit_cursor_boundary) =
                     select_pending(&tasks, &state, &mut newest_updated);
 
+                // ctx.sync_depth_days: drop items updated before the depth floor. `pending` is
+                // in descending timestamp order, so truncate at the first item below the floor
+                // and signal cursor-boundary so pagination stops.
+                if let Some(ref floor) = oldest_allowed_time {
+                    if let Some(cut) = pending.iter().position(|p| {
+                        p.updated
+                            .as_deref()
+                            .map(|t| t < floor.as_str())
+                            .unwrap_or(false)
+                    }) {
+                        pending.truncate(cut);
+                        hit_cursor_boundary = true;
+                    }
+                }
+
+                // ctx.max_items: clamp the dedup'd batch to the remaining budget before ingest.
+                cap.clamp_batch(&mut pending);
+
                 let ingestor = MemoryTreeIngestor {
                     config: ctx.config.as_ref(),
                     connection_id: &connection_id,
@@ -341,6 +389,13 @@ impl ComposioProvider for ClickUpProvider {
                     state.mark_synced(key);
                 }
                 total_persisted += outcome.persisted;
+                cap.record(outcome.persisted);
+
+                // ctx.max_items precise cap: once the per-source cap is hit, stop paginating.
+                if cap.is_reached() {
+                    hit_cap_boundary = true;
+                    break 'workspaces;
+                }
 
                 if hit_cursor_boundary {
                     tracing::debug!(
@@ -351,6 +406,18 @@ impl ComposioProvider for ClickUpProvider {
                     break;
                 }
 
+                // ctx.max_items hard stop.
+                if cap.is_reached() {
+                    tracing::debug!(
+                        workspace_id = %workspace_id,
+                        page = page_num,
+                        total_persisted,
+                        "[composio:clickup] [memory_sync] max_items reached, stopping pagination"
+                    );
+                    hit_cap_boundary = true;
+                    break 'workspaces;
+                }
+
                 // ClickUp's filtered-team-tasks endpoint signals the last
                 // page implicitly: when fewer than `page_size` results
                 // come back, there are no more pages.
@@ -367,8 +434,17 @@ impl ComposioProvider for ClickUpProvider {
         }
 
         // ── Step 6: advance cursor and save state ───────────────────
-        if let Some(new_cursor) = newest_updated {
-            state.advance_cursor(&new_cursor);
+        // Hold the cursor on a cap-truncated pass so the next sync re-scans the unseen tail.
+        if !hit_cap_boundary {
+            if let Some(new_cursor) = newest_updated {
+                state.advance_cursor(&new_cursor);
+            }
+        } else {
+            tracing::warn!(
+                connection_id = %connection_id,
+                "[composio:clickup] holding cursor — cap-truncated pass; next sync will re-scan \
+                 the unseen tail"
+            );
         }
         state.set_last_sync_at_ms(sync::now_ms());
         state.save(&memory).await?;

src/openhuman/memory_sync/composio/providers/github/provider.rs

@@ -200,8 +200,42 @@ impl ComposioProvider for GitHubProvider {
             _ => PAGE_SIZE,
         };
 
-        // Build the base search query.
-        let query = build_search_query(&login, state.cursor.as_deref());
+        // ctx.max_items: route through ItemCap — page ceiling, mid-page
+        // per-item break, and post-page hard stop all share one source of truth.
+        let mut cap = super::super::helpers::ItemCap::new(ctx.max_items);
+        let effective_max_pages = cap.max_pages(page_size, MAX_PAGES);
+        if ctx.max_items.is_some() && effective_max_pages < MAX_PAGES {
+            tracing::debug!(
+                connection_id = %connection_id,
+                max_items = ?ctx.max_items,
+                effective_max_pages,
+                "[composio:github] [memory_sync] applying max_items page cap"
+            );
+        }
+
+        // ctx.sync_depth_days: inject `updated:>{date}` on first sync when set.
+        let depth_query_fragment: Option<String> = if state.cursor.is_none() {
+            ctx.sync_depth_days.map(|days| {
+                let floor = chrono::Utc::now() - chrono::Duration::days(days as i64);
+                let s = floor.format("%Y-%m-%dT%H:%M:%SZ").to_string();
+                tracing::debug!(
+                    connection_id = %connection_id,
+                    sync_depth_days = days,
+                    floor = %s,
+                    "[composio:github] [memory_sync] injecting updated:> date filter on first sync"
+                );
+                format!("updated:>{s}")
+            })
+        } else {
+            None
+        };
+
+        // Build the base search query — include depth fragment when set.
+        let query = build_search_query_with_depth(
+            &login,
+            state.cursor.as_deref(),
+            depth_query_fragment.as_deref(),
+        );
 
         let mut total_fetched: usize = 0;
         let mut total_persisted: usize = 0;
@@ -214,8 +248,9 @@ impl ComposioProvider for GitHubProvider {
         // until its next edit. Already-synced items are skipped cheaply via
         // `is_synced` on the re-fetch, so the cost of holding is minimal.
         let mut had_ingest_failures = false;
+        let mut hit_cap_boundary = false;
 
-        'pages: for page_num in 1..=MAX_PAGES {
+        'pages: for page_num in 1..=effective_max_pages {
             if state.budget_exhausted() {
                 tracing::info!(
                     page = page_num,
@@ -335,6 +370,7 @@ impl ComposioProvider for GitHubProvider {
                     Ok(_chunks_written) => {
                         state.mark_synced(&sync_key);
                         total_persisted += 1;
+                        cap.record(1);
                     }
                     Err(e) => {
                         had_ingest_failures = true;
@@ -345,6 +381,24 @@ impl ComposioProvider for GitHubProvider {
                         );
                     }
                 }
+
+                // ctx.max_items precise cap: stop mid-page so we never persist
+                // more than the cap even when a single page exceeds it.
+                if cap.is_reached() {
+                    hit_cap_boundary = true;
+                    break;
+                }
+            }
+
+            // ctx.max_items hard stop.
+            if cap.is_reached() {
+                tracing::debug!(
+                    page = page_num,
+                    total_persisted,
+                    "[composio:github] [memory_sync] max_items reached, stopping pagination"
+                );
+                hit_cap_boundary = true;
+                break;
             }
 
             // GitHub search pages are 0-indexed in terms of total results;
@@ -366,15 +420,18 @@ impl ComposioProvider for GitHubProvider {
         // the delete-first memory-tree pipeline (#2885). `set_last_sync_at_ms`
         // still advances — that's just a heartbeat, not a fetch-window
         // boundary, so it's safe to record that we did attempt a sync.
-        if !had_ingest_failures {
+        // Hold the cursor on a cap-truncated pass so the next sync re-scans the unseen tail.
+        if !had_ingest_failures && !hit_cap_boundary {
             if let Some(new_cursor) = newest_updated {
                 state.advance_cursor(&new_cursor);
             }
         } else {
             tracing::warn!(
                 connection_id = %connection_id,
-                "[composio:github] holding cursor — ingest failures this pass; next sync will \
-                 re-fetch the failed range"
+                had_ingest_failures,
+                hit_cap_boundary,
+                "[composio:github] holding cursor — ingest failures or cap-truncated pass; next \
+                 sync will re-fetch the failed range"
             );
         }
         state.set_last_sync_at_ms(sync::now_ms());
@@ -896,3 +953,48 @@ pub(super) fn build_search_query(login: &str, cursor: Option<&str>) -> String {
         None => format!("involves:{login}"),
     }
 }
+
+/// Extended variant that optionally appends a `sync_depth_days` fragment on
+/// first sync (no cursor). The `depth_fragment` is expected to be a pre-built
+/// `"updated:>{date}"` string.
+pub(super) fn build_search_query_with_depth(
+    login: &str,
+    cursor: Option<&str>,
+    depth_fragment: Option<&str>,
+) -> String {
+    match cursor {
+        Some(c) => format!("involves:{login} updated:>{c}"),
+        None => match depth_fragment {
+            Some(fragment) => format!("involves:{login} {fragment}"),
+            None => format!("involves:{login}"),
+        },
+    }
+}
+
+#[cfg(test)]
+mod depth_tests {
+    use super::*;
+
+    #[test]
+    fn build_search_query_with_depth_no_cursor_no_depth() {
+        let q = build_search_query_with_depth("alice", None, None);
+        assert_eq!(q, "involves:alice");
+    }
+
+    #[test]
+    fn build_search_query_with_depth_no_cursor_with_depth() {
+        let q = build_search_query_with_depth("alice", None, Some("updated:>2024-01-01T00:00:00Z"));
+        assert_eq!(q, "involves:alice updated:>2024-01-01T00:00:00Z");
+    }
+
+    #[test]
+    fn build_search_query_with_depth_cursor_wins_over_depth() {
+        // When cursor is set, depth fragment is ignored.
+        let q = build_search_query_with_depth(
+            "alice",
+            Some("2024-06-01T00:00:00Z"),
+            Some("updated:>2024-01-01T00:00:00Z"),
+        );
+        assert_eq!(q, "involves:alice updated:>2024-06-01T00:00:00Z");
+    }
+}

src/openhuman/memory_sync/composio/providers/gmail/provider.rs

@@ -258,7 +258,7 @@ impl ComposioProvider for GmailProvider {
         // legitimately want the larger ceiling — and the cap only
         // kicks in when we have a prior `last_sync_at_ms` to compare
         // against, so first-ever syncs are unaffected.
-        let max_pages = match reason {
+        let base_max_pages = match reason {
             SyncReason::ConnectionCreated => MAX_PAGES_PER_SYNC,
             _ => match state.last_sync_at_ms {
                 Some(last_ms) if sync::now_ms().saturating_sub(last_ms) < RECENT_SYNC_WINDOW_MS => {
@@ -274,6 +274,36 @@ impl ComposioProvider for GmailProvider {
             },
         };
 
+        // ctx.max_items: route through ItemCap so the page ceiling, mid-page
+        // clamp, and post-page hard stop all share one source of truth.
+        let mut cap = super::super::helpers::ItemCap::new(ctx.max_items);
+        let max_pages = cap.max_pages(page_size, base_max_pages);
+        if ctx.max_items.is_some() && max_pages < base_max_pages {
+            tracing::debug!(
+                connection_id = %connection_id,
+                max_items = ?ctx.max_items,
+                page_size,
+                effective_max_pages = max_pages,
+                "[composio:gmail] [memory_sync] applying max_items page cap from source config"
+            );
+        }
+
+        // ctx.sync_depth_days: on first sync (no cursor), add an after:<epoch> floor.
+        let depth_floor_filter: Option<String> = if state.cursor.is_none() {
+            ctx.sync_depth_days.map(|days| {
+                let floor_secs = super::super::helpers::epoch_floor_from_depth(days);
+                tracing::debug!(
+                    connection_id = %connection_id,
+                    sync_depth_days = days,
+                    floor_epoch_secs = floor_secs,
+                    "[composio:gmail] [memory_sync] applying sync_depth_days floor on first sync"
+                );
+                floor_secs.to_string()
+            })
+        } else {
+            None
+        };
+
         let mut total_fetched: usize = 0;
         let mut total_persisted: usize = 0;
         let mut total_requests: u32 = 0;
@@ -281,6 +311,7 @@ impl ComposioProvider for GmailProvider {
         let mut newest_id: Option<String> = None;
         let mut page_token: Option<String> = None;
         let mut stop_reason: &'static str = "max_pages";
+        let mut hit_cap_boundary = false;
 
         for page_num in 0..max_pages {
             if state.budget_exhausted() {
@@ -321,6 +352,9 @@ impl ComposioProvider for GmailProvider {
                         "[composio:gmail] using day-level filter from cursor (epoch parse failed)"
                     );
                 }
+            } else if let Some(ref floor) = depth_floor_filter {
+                // First sync with sync_depth_days: apply the epoch floor.
+                query.push_str(&format!(" after:{floor}"));
             }
 
             let mut args = json!({
@@ -453,6 +487,11 @@ impl ComposioProvider for GmailProvider {
                 new_messages.push(msg.clone());
             }
 
+            // ctx.max_items precise cap: clamp the per-page batch before ingest
+            // so a single page larger than the budget is never over-persisted.
+            cap.clamp_batch(&mut new_messages);
+            cap.clamp_batch(&mut pending_synced_ids);
+
             // Single batched ingest into memory_tree. Chunk IDs are
             // content-hashed so re-ingest of the same message is an
             // idempotent UPSERT at the SQL layer; per-message dedup above
@@ -483,6 +522,7 @@ impl ComposioProvider for GmailProvider {
                         // persist path. n is the chunk count which we log
                         // for diagnostic purposes only.
                         total_persisted += new_messages.len();
+                        cap.record(new_messages.len());
                         tracing::debug!(
                             page = page_num,
                             new_messages = new_messages.len(),
@@ -512,6 +552,18 @@ impl ComposioProvider for GmailProvider {
                 break;
             }
 
+            // ctx.max_items hard stop: break once the per-source cap is reached.
+            if cap.is_reached() {
+                tracing::debug!(
+                    page = page_num,
+                    total_persisted,
+                    "[composio:gmail] [memory_sync] max_items reached, stopping pagination"
+                );
+                stop_reason = "max_items";
+                hit_cap_boundary = true;
+                break;
+            }
+
             // Check for next page token.
             page_token = sync::extract_page_token(&resp.data);
             if page_token.is_none() {
@@ -522,8 +574,17 @@ impl ComposioProvider for GmailProvider {
         }
 
         // ── Step 5: advance cursor and save state ───────────────────
-        if let Some(new_cursor) = newest_date {
-            state.advance_cursor(&new_cursor);
+        // Hold the cursor on a cap-truncated pass so the next sync re-scans the unseen tail.
+        if !hit_cap_boundary {
+            if let Some(new_cursor) = newest_date {
+                state.advance_cursor(&new_cursor);
+            }
+        } else {
+            tracing::warn!(
+                connection_id = %connection_id,
+                "[composio:gmail] holding cursor — cap-truncated pass; next sync will re-scan \
+                 the unseen tail"
+            );
         }
         if let Some(ref freshest) = newest_id {
             state.set_last_seen_id(freshest);
@@ -616,3 +677,7 @@ impl ComposioProvider for GmailProvider {
         Ok(())
     }
 }
+
+// Cap/date-floor math lives in the shared `super::super::helpers` module
+// (`ItemCap`, `pages_for_max_items`, `epoch_floor_from_depth`) so every provider
+// shares one implementation — see that module for the unit tests.

src/openhuman/memory_sync/composio/providers/helpers.rs

@@ -46,6 +46,178 @@ pub(crate) fn merge_extra(args: &mut serde_json::Value, extra: &serde_json::Valu
     }
 }
 
+// ── Cap enforcement helpers ──────────────────────────────────────────────
+
+/// Compute the number of API pages needed to cover `max_items` at `page_size`
+/// items per page, rounding up.
+///
+/// Returns `u32::MAX` when `page_size == 0` to avoid division by zero;
+/// callers should treat this as "no page cap beyond the provider's own upper
+/// bound".
+pub(crate) fn pages_for_max_items(max_items: u32, page_size: u32) -> u32 {
+    if page_size == 0 {
+        return u32::MAX;
+    }
+    // Widen to u64 before the addition to prevent overflow for large cap values.
+    (((max_items as u64) + (page_size as u64) - 1) / (page_size as u64)).min(u32::MAX as u64) as u32
+}
+
+/// Compute the Unix epoch timestamp (seconds) for `sync_depth_days` days ago.
+/// Used to build after-date filters (e.g. Gmail `after:<epoch>`) on first sync.
+pub(crate) fn epoch_floor_from_depth(sync_depth_days: u32) -> i64 {
+    let now = chrono::Utc::now();
+    let floor = now - chrono::Duration::days(sync_depth_days as i64);
+    floor.timestamp()
+}
+
+/// Single source of truth for the per-sync `max_items` cap.
+///
+/// Every Composio provider used to open-code three near-identical blocks — a
+/// page-count cap, a mid-page clamp, and a post-page hard stop — which is how
+/// the same off-by-a-page bug ended up in five providers and was missed in a
+/// sixth. Funnelling all of them through this one type keeps the rule in one
+/// place: construct it from `ctx.max_items`, derive the page cap, clamp each
+/// batch (or check per item), and stop once the cap is reached.
+///
+/// `None` cap means "no item limit beyond the provider's own internal page
+/// ceiling" (e.g. after the user clicks "All In").
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct ItemCap {
+    cap: Option<usize>,
+    persisted: usize,
+}
+
+impl ItemCap {
+    /// Build from a source's `max_items` value (`None` = uncapped).
+    pub(crate) fn new(max_items: Option<u32>) -> Self {
+        Self {
+            cap: max_items.map(|n| n as usize),
+            persisted: 0,
+        }
+    }
+
+    /// The page ceiling to actually fetch: the smaller of the provider's own
+    /// `fallback` (e.g. `MAX_PAGES_PER_SYNC`) and the pages needed to cover the
+    /// cap. Uncapped → `fallback` unchanged.
+    pub(crate) fn max_pages(&self, page_size: u32, fallback: u32) -> u32 {
+        match self.cap {
+            Some(cap) => pages_for_max_items(cap as u32, page_size).min(fallback),
+            None => fallback,
+        }
+    }
+
+    /// How many more items may still be persisted. `None` = unlimited.
+    pub(crate) fn remaining(&self) -> Option<usize> {
+        self.cap.map(|cap| cap.saturating_sub(self.persisted))
+    }
+
+    /// True once the cap is set and reached — callers `break` their pagination.
+    pub(crate) fn is_reached(&self) -> bool {
+        matches!(self.remaining(), Some(0))
+    }
+
+    /// Record `n` newly-persisted items against the budget.
+    pub(crate) fn record(&mut self, n: usize) {
+        self.persisted = self.persisted.saturating_add(n);
+    }
+
+    /// Truncate a to-ingest batch down to the remaining budget, so a single
+    /// page larger than the cap never over-persists. No-op when uncapped.
+    pub(crate) fn clamp_batch<T>(&self, batch: &mut Vec<T>) {
+        if let Some(remaining) = self.remaining() {
+            if batch.len() > remaining {
+                batch.truncate(remaining);
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod cap_helper_tests {
+    use super::*;
+
+    #[test]
+    fn pages_for_max_items_rounds_up() {
+        assert_eq!(pages_for_max_items(100, 25), 4);
+        assert_eq!(pages_for_max_items(101, 25), 5);
+        assert_eq!(pages_for_max_items(1, 25), 1);
+        assert_eq!(pages_for_max_items(50, 50), 1);
+        assert_eq!(pages_for_max_items(51, 50), 2);
+    }
+
+    #[test]
+    fn pages_for_max_items_zero_page_size() {
+        assert_eq!(pages_for_max_items(100, 0), u32::MAX);
+    }
+
+    #[test]
+    fn epoch_floor_from_depth_is_in_the_past() {
+        let floor = epoch_floor_from_depth(30);
+        let now = chrono::Utc::now().timestamp();
+        assert!(floor < now);
+        let diff_days = (now - floor) / 86400;
+        assert!(
+            diff_days >= 29 && diff_days <= 31,
+            "expected ~30 days in past, got {diff_days}"
+        );
+    }
+
+    #[test]
+    fn item_cap_uncapped_is_never_reached() {
+        let mut cap = ItemCap::new(None);
+        assert_eq!(cap.remaining(), None);
+        assert!(!cap.is_reached());
+        cap.record(1_000_000);
+        assert!(!cap.is_reached());
+        assert_eq!(
+            cap.max_pages(25, 20),
+            20,
+            "uncapped keeps the provider fallback"
+        );
+    }
+
+    #[test]
+    fn item_cap_tracks_remaining_and_reached() {
+        let mut cap = ItemCap::new(Some(3));
+        assert_eq!(cap.remaining(), Some(3));
+        assert!(!cap.is_reached());
+        cap.record(2);
+        assert_eq!(cap.remaining(), Some(1));
+        assert!(!cap.is_reached());
+        cap.record(5); // saturates, never underflows
+        assert_eq!(cap.remaining(), Some(0));
+        assert!(cap.is_reached());
+    }
+
+    #[test]
+    fn item_cap_max_pages_is_min_of_fallback_and_needed() {
+        // cap=2, page_size=50 → 1 page needed, well under the fallback.
+        assert_eq!(ItemCap::new(Some(2)).max_pages(50, 20), 1);
+        // cap=1000, page_size=25 → 40 pages needed, clamped to fallback 20.
+        assert_eq!(ItemCap::new(Some(1000)).max_pages(25, 20), 20);
+    }
+
+    #[test]
+    fn item_cap_clamp_batch_truncates_to_remaining() {
+        let cap = ItemCap::new(Some(2));
+        let mut batch = vec![1, 2, 3, 4, 5];
+        cap.clamp_batch(&mut batch);
+        assert_eq!(batch, vec![1, 2]);
+
+        // Uncapped leaves the batch untouched.
+        let mut full = vec![1, 2, 3];
+        ItemCap::new(None).clamp_batch(&mut full);
+        assert_eq!(full, vec![1, 2, 3]);
+
+        // After recording progress, clamp uses the reduced budget.
+        let mut cap2 = ItemCap::new(Some(5));
+        cap2.record(3);
+        let mut batch2 = vec![1, 2, 3, 4];
+        cap2.clamp_batch(&mut batch2);
+        assert_eq!(batch2, vec![1, 2], "only 2 of the 5 budget remained");
+    }
+}
+
 /// Resolve the first array found among `array_paths` (dotted object
 /// paths), then return the first non-empty string at one of `fields`
 /// on that array's first element. Complements [`pick_str`], which

src/openhuman/memory_sync/composio/providers/linear/provider.rs

@@ -198,14 +198,41 @@ impl ComposioProvider for LinearProvider {
             _ => PAGE_SIZE,
         };
 
+        // ctx.max_items: route through ItemCap — page ceiling, mid-page
+        // per-item break, and post-page hard stop all share one source of truth.
+        let mut cap = super::super::helpers::ItemCap::new(ctx.max_items);
+        let effective_max_pages = cap.max_pages(page_size as u32, MAX_PAGES_PER_SYNC);
+        if ctx.max_items.is_some() && effective_max_pages < MAX_PAGES_PER_SYNC {
+            tracing::debug!(
+                connection_id = %connection_id,
+                max_items = ?ctx.max_items,
+                effective_max_pages,
+                "[composio:linear] [memory_sync] applying max_items page cap"
+            );
+        }
+
+        // ctx.sync_depth_days: oldest allowed updatedAt for client-side skip.
+        let oldest_allowed_time: Option<String> = ctx.sync_depth_days.map(|days| {
+            let floor = chrono::Utc::now() - chrono::Duration::days(days as i64);
+            let s = floor.to_rfc3339();
+            tracing::debug!(
+                connection_id = %connection_id,
+                sync_depth_days = days,
+                oldest_allowed = %s,
+                "[composio:linear] [memory_sync] applying sync_depth_days floor"
+            );
+            s
+        });
+
         let mut total_fetched: usize = 0;
         let mut total_persisted: usize = 0;
         let mut had_persist_failures = false;
         let mut newest_updated: Option<String> = None;
         let mut after_cursor: Option<String> = None;
         let mut hit_cursor_boundary = false;
+        let mut hit_cap_boundary = false;
 
-        for page_num in 0..MAX_PAGES_PER_SYNC {
+        for page_num in 0..effective_max_pages {
             if state.budget_exhausted() {
                 tracing::info!(
                     page = page_num,
@@ -256,11 +283,30 @@ impl ComposioProvider for LinearProvider {
             }
 
             // ── Per-item dedup + bounded-concurrency ingest ──────────
-            let (pending, page_hit_boundary) = select_pending(&issues, &state, &mut newest_updated);
+            let (mut pending, page_hit_boundary) =
+                select_pending(&issues, &state, &mut newest_updated);
             if page_hit_boundary {
                 hit_cursor_boundary = true;
             }
 
+            // ctx.sync_depth_days: drop items updated before the depth floor. `pending` is
+            // in descending timestamp order, so truncate at the first item below the floor
+            // and signal cursor-boundary so pagination stops.
+            if let Some(ref floor) = oldest_allowed_time {
+                if let Some(cut) = pending.iter().position(|p| {
+                    p.updated
+                        .as_deref()
+                        .map(|t| t < floor.as_str())
+                        .unwrap_or(false)
+                }) {
+                    pending.truncate(cut);
+                    hit_cursor_boundary = true;
+                }
+            }
+
+            // ctx.max_items: clamp the dedup'd batch to the remaining budget before ingest.
+            cap.clamp_batch(&mut pending);
+
             let ingestor = MemoryTreeIngestor {
                 config: ctx.config.as_ref(),
                 connection_id: &connection_id,
@@ -270,10 +316,17 @@ impl ComposioProvider for LinearProvider {
                 state.mark_synced(key);
             }
             total_persisted += outcome.persisted;
+            cap.record(outcome.persisted);
             if outcome.had_failures {
                 had_persist_failures = true;
             }
 
+            // ctx.max_items precise cap: once the per-source cap is hit, stop paginating.
+            if cap.is_reached() {
+                hit_cap_boundary = true;
+                break;
+            }
+
             if hit_cursor_boundary {
                 tracing::debug!(
                     page = page_num,
@@ -282,6 +335,17 @@ impl ComposioProvider for LinearProvider {
                 break;
             }
 
+            // ctx.max_items hard stop.
+            if cap.is_reached() {
+                tracing::debug!(
+                    page = page_num,
+                    total_persisted,
+                    "[composio:linear] [memory_sync] max_items reached, stopping pagination"
+                );
+                hit_cap_boundary = true;
+                break;
+            }
+
             // Advance to the next page using Linear's cursor-based pagination.
             match sync::extract_pagination_cursor(&resp.data) {
                 Some(next_cursor) => {
@@ -298,10 +362,17 @@ impl ComposioProvider for LinearProvider {
         }
 
         // ── Step 5: advance cursor and save state ────────────────────
+        // Hold the cursor on a cap-truncated pass so the next sync re-scans the unseen tail.
         if had_persist_failures {
             tracing::warn!(
                 "[composio:linear] persist failures seen; keeping previous cursor for retry"
             );
+        } else if hit_cap_boundary {
+            tracing::warn!(
+                hit_cap_boundary,
+                "[composio:linear] holding cursor — cap-truncated pass; next sync will re-scan \
+                 the unseen tail"
+            );
         } else if let Some(new_cursor) = newest_updated {
             state.advance_cursor(&new_cursor);
         }

src/openhuman/memory_sync/composio/providers/notion/provider.rs

@@ -186,6 +186,33 @@ impl ComposioProvider for NotionProvider {
             _ => PAGE_SIZE,
         };
 
+        // ctx.max_items: route through ItemCap — page ceiling, mid-page
+        // per-item break, and post-page hard stop all share one source of truth.
+        let mut cap = super::super::helpers::ItemCap::new(ctx.max_items);
+        let effective_max_pages = cap.max_pages(page_size, MAX_PAGES_PER_SYNC);
+        if ctx.max_items.is_some() && effective_max_pages < MAX_PAGES_PER_SYNC {
+            tracing::debug!(
+                connection_id = %connection_id,
+                max_items = ?ctx.max_items,
+                effective_max_pages,
+                "[composio:notion] [memory_sync] applying max_items page cap"
+            );
+        }
+
+        // ctx.sync_depth_days: compute the oldest allowed edited_time string
+        // for client-side skipping of stale results.
+        let oldest_allowed_time: Option<String> = ctx.sync_depth_days.map(|days| {
+            let floor = chrono::Utc::now() - chrono::Duration::days(days as i64);
+            let s = floor.to_rfc3339();
+            tracing::debug!(
+                connection_id = %connection_id,
+                sync_depth_days = days,
+                oldest_allowed = %s,
+                "[composio:notion] [memory_sync] applying sync_depth_days floor"
+            );
+            s
+        });
+
         let mut total_fetched: usize = 0;
         let mut total_persisted: usize = 0;
         let mut newest_edited_time: Option<String> = None;
@@ -198,8 +225,9 @@ impl ComposioProvider for NotionProvider {
         // its next edit. Already-synced items are skipped cheaply via
         // `is_synced` on the re-fetch, so the cost of holding is minimal.
         let mut had_ingest_failures = false;
+        let mut hit_cap_boundary = false;
 
-        for page_num in 0..MAX_PAGES_PER_SYNC {
+        for page_num in 0..effective_max_pages {
             if state.budget_exhausted() {
                 tracing::info!(
                     page = page_num,
@@ -249,9 +277,27 @@ impl ComposioProvider for NotionProvider {
             }
 
             // ── Step 4a: dedupe + decide which pages to ingest ──────
-            let (pending, hit_cursor_boundary) =
+            let (mut pending, mut hit_cursor_boundary) =
                 select_pending(&results, &state, &mut newest_edited_time);
 
+            // ctx.sync_depth_days: drop items edited before the depth floor. `pending` is
+            // in descending timestamp order, so truncate at the first item below the floor
+            // and signal cursor-boundary so pagination stops.
+            if let Some(ref floor) = oldest_allowed_time {
+                if let Some(cut) = pending.iter().position(|p| {
+                    p.edited_time
+                        .as_deref()
+                        .map(|t| t < floor.as_str())
+                        .unwrap_or(false)
+                }) {
+                    pending.truncate(cut);
+                    hit_cursor_boundary = true;
+                }
+            }
+
+            // ctx.max_items: clamp the dedup'd batch to the remaining budget before ingest.
+            cap.clamp_batch(&mut pending);
+
             // ── Step 4b: ingest queued pages (bounded concurrency) ──
             let ingestor = MemoryTreeIngestor {
                 config: ctx.config.as_ref(),
@@ -262,10 +308,17 @@ impl ComposioProvider for NotionProvider {
                 state.mark_synced(key);
             }
             total_persisted += outcome.persisted;
+            cap.record(outcome.persisted);
             if outcome.had_failures {
                 had_ingest_failures = true;
             }
 
+            // ctx.max_items precise cap: once the per-source cap is hit, stop paginating.
+            if cap.is_reached() {
+                hit_cap_boundary = true;
+                break;
+            }
+
             if hit_cursor_boundary {
                 tracing::debug!(
                     page = page_num,
@@ -274,6 +327,17 @@ impl ComposioProvider for NotionProvider {
                 break;
             }
 
+            // ctx.max_items hard stop.
+            if cap.is_reached() {
+                tracing::debug!(
+                    page = page_num,
+                    total_persisted,
+                    "[composio:notion] [memory_sync] max_items reached, stopping pagination"
+                );
+                hit_cap_boundary = true;
+                break;
+            }
+
             // Check for next page cursor from Notion API.
             notion_cursor = sync::extract_notion_cursor(&resp.data);
             if notion_cursor.is_none() {
@@ -287,15 +351,18 @@ impl ComposioProvider for NotionProvider {
         // Hold the cursor when any item failed to ingest this pass. See the
         // `had_ingest_failures` declaration above for why this matters under
         // the delete-first memory-tree pipeline (#2885).
-        if !had_ingest_failures {
+        // Hold the cursor on a cap-truncated pass so the next sync re-scans the unseen tail.
+        if !had_ingest_failures && !hit_cap_boundary {
             if let Some(new_cursor) = newest_edited_time {
                 state.advance_cursor(&new_cursor);
             }
         } else {
             tracing::warn!(
                 connection_id = %connection_id,
-                "[composio:notion] holding cursor — ingest failures this pass; next sync will \
-                 re-fetch the failed range"
+                had_ingest_failures,
+                hit_cap_boundary,
+                "[composio:notion] holding cursor — ingest failures or cap-truncated pass; next \
+                 sync will re-fetch the failed range"
             );
         }
         state.save(&memory).await?;

src/openhuman/memory_sync/composio/providers/slack/provider.rs

@@ -468,6 +468,13 @@ impl ComposioProvider for SlackProvider {
         let mut total_messages_ingested: usize = 0;
         let mut channels_processed: usize = 0;
         let mut channels_errored: usize = 0;
+        let mut hit_cap_boundary = false;
+
+        // ctx.max_items: ItemCap is threaded through process_channel so the
+        // per-page batch is clamped before ingest and the channel loop stops
+        // precisely at the cap — the old coarse post-channel check allowed a
+        // single page/channel to blow past the cap.
+        let mut cap = super::super::helpers::ItemCap::new(ctx.max_items);
 
         // 2. Per-channel: fetch → post-process → enrich → ingest.
         for channel in &channels {
@@ -488,6 +495,7 @@ impl ComposioProvider for SlackProvider {
                 now,
                 &users,
                 &connection_id,
+                &mut cap,
             )
             .await
             {
@@ -506,6 +514,27 @@ impl ComposioProvider for SlackProvider {
                 }
             }
 
+            // ctx.max_items hard stop across all channels (precise — cap was
+            // already applied inside process_channel so this break fires
+            // exactly when the budget is exhausted, not one channel later).
+            if cap.is_reached() {
+                // Hold the cursor on a cap-truncated pass so the next sync re-scans the unseen tail.
+                hit_cap_boundary = true;
+                tracing::debug!(
+                    connection_id = %connection_id,
+                    total_messages_ingested,
+                    "[composio:slack] [memory_sync] max_items reached, stopping channel iteration"
+                );
+                // Save state before breaking without advancing the cursor.
+                if let Err(err) = state.save(&memory).await {
+                    tracing::warn!(
+                        error = %err,
+                        "[composio:slack] state save failed after cap-stop (non-fatal)"
+                    );
+                }
+                break;
+            }
+
             state.advance_cursor(sync::encode_cursors(&cursors));
             if let Err(err) = state.save(&memory).await {
                 tracing::warn!(
@@ -515,6 +544,15 @@ impl ComposioProvider for SlackProvider {
             }
         }
 
+        if hit_cap_boundary {
+            // Hold the cursor on a cap-truncated pass so the next sync re-scans the unseen tail.
+            tracing::warn!(
+                connection_id = %connection_id,
+                "[composio:slack] cap-truncated pass; cursor held so next sync re-scans the \
+                 unseen tail"
+            );
+        }
+
         let finished_at_ms = sync::now_ms();
         let summary = format!(
             "slack sync: channels_processed={channels_processed} \
@@ -608,7 +646,12 @@ async fn list_all_channels(
 }
 
 /// Pull one channel's history since its cursor, post-process + enrich each
-/// page, then ingest all messages. Returns the number of chunks written.
+/// page, then ingest all messages. Returns the number of messages written.
+///
+/// `cap` is the shared [`super::super::helpers::ItemCap`] for the sync pass.
+/// Each page's message batch is clamped to the remaining budget before ingest
+/// so the per-sync `max_items` limit is respected precisely regardless of how
+/// many messages a single channel/page returns.
 async fn process_channel(
     ctx: &ProviderContext,
     state: &mut SyncState,
@@ -617,13 +660,26 @@ async fn process_channel(
     now: chrono::DateTime<chrono::Utc>,
     users: &SlackUsers,
     connection_id: &str,
+    cap: &mut super::super::helpers::ItemCap,
 ) -> Result<usize, String> {
     // Cursor value is a raw Slack `ts` (`"<seconds>.<micro>"`) preserved
     // with full precision, so multi-message-per-second channels don't
     // replay the whole second on the next incremental fetch. When no
     // cursor exists yet, fall back to `<backfill_window_secs>.000000`.
+    // ctx.sync_depth_days wins over the env-var OPENHUMAN_SLACK_BACKFILL_DAYS
+    // default when set — it comes from the user-configured source entry.
     let oldest_ts = cursors.get(&channel.id).cloned().unwrap_or_else(|| {
-        let secs = (now - chrono::Duration::days(backfill_days())).timestamp();
+        let depth_days = ctx
+            .sync_depth_days
+            .map(|d| d as i64)
+            .unwrap_or_else(backfill_days);
+        let secs = (now - chrono::Duration::days(depth_days)).timestamp();
+        tracing::debug!(
+            channel = %channel.id,
+            depth_days,
+            oldest_ts_secs = secs,
+            "[composio:slack] [memory_sync] computing oldest_ts for backfill"
+        );
         format!("{secs}.000000")
     });
 
@@ -676,6 +732,23 @@ async fn process_channel(
             break;
         }
         all_messages.extend(msgs);
+
+        // Stop fetching further pages for this channel if we have already
+        // accumulated enough to fill the remaining budget (checked against
+        // remaining() which accounts for items recorded by previous channels).
+        if let Some(remaining) = cap.remaining() {
+            if all_messages.len() >= remaining {
+                tracing::debug!(
+                    channel = %channel.id,
+                    page = page_num,
+                    accumulated = all_messages.len(),
+                    remaining,
+                    "[composio:slack] [memory_sync] budget nearly full, stopping history pagination"
+                );
+                break;
+            }
+        }
+
         cursor = sync::extract_next_cursor(&resp.data);
         if cursor.is_none() {
             break;
@@ -690,6 +763,19 @@ async fn process_channel(
         return Ok(0);
     }
 
+    // ctx.max_items precise cap: clamp the full accumulated batch to the
+    // remaining budget before ingest so we never persist more than the cap
+    // allows, even if a single channel/page returned more than what remains.
+    cap.clamp_batch(&mut all_messages);
+
+    if all_messages.is_empty() {
+        tracing::debug!(
+            channel = %channel.id,
+            "[composio:slack] [memory_sync] cap already reached, skipping channel ingest"
+        );
+        return Ok(0);
+    }
+
     let msg_count = all_messages.len();
     tracing::info!(
         channel = %channel.id,
@@ -713,13 +799,16 @@ async fn process_channel(
             {
                 cursors.insert(channel.id.clone(), latest);
             }
+            cap.record(msg_count);
             tracing::info!(
                 channel = %channel.id,
                 messages = msg_count,
                 chunks,
                 "[composio:slack] channel ingest done"
             );
-            Ok(chunks)
+            // Return message count (consistent with the sync path which
+            // counts messages, not chunks, for the items_ingested metric).
+            Ok(msg_count)
         }
         Err(e) => {
             tracing::warn!(

src/openhuman/memory_sync/composio/providers/slack/rpc.rs

@@ -111,6 +111,8 @@ pub async fn sync_trigger_rpc(
             toolkit: conn.toolkit.clone(),
             connection_id: Some(conn.id.clone()),
             usage: Default::default(),
+            max_items: None,
+            sync_depth_days: None,
         };
         match provider.sync(&ctx, SyncReason::Manual).await {
             Ok(o) => outcomes.push(o),

src/openhuman/memory_sync/composio/providers/types.rs

@@ -295,6 +295,18 @@ pub struct ProviderContext {
     /// (`run_connection_sync`) reads it back. Non-sync callers (agent tools,
     /// task-source fetches) leave it at zero — harmless.
     pub usage: ComposioUsageHandle,
+    /// Maximum items to fetch in a single sync pass.
+    ///
+    /// Set from the corresponding `MemorySourceEntry.max_items` field at
+    /// sync-dispatch time. `None` means no cap beyond the provider's own
+    /// internal upper bounds.
+    pub max_items: Option<u32>,
+    /// Maximum sync depth window in days.
+    ///
+    /// Set from `MemorySourceEntry.sync_depth_days`. When `Some(n)`, the
+    /// provider only fetches items from the last `n` days. `None` means
+    /// no additional depth restriction beyond the provider's cursor.
+    pub sync_depth_days: Option<u32>,
 }
 
 impl ProviderContext {
@@ -325,6 +337,8 @@ impl ProviderContext {
                 toolkit: toolkit.into(),
                 connection_id,
                 usage: ComposioUsageHandle::default(),
+                max_items: None,
+                sync_depth_days: None,
             }),
             Err(e) => {
                 tracing::debug!(
@@ -487,6 +501,8 @@ mod tests {
             toolkit: "gmail".to_string(),
             connection_id: None,
             usage: ComposioUsageHandle::default(),
+            max_items: None,
+            sync_depth_days: None,
         };
         let cloned = ctx.clone();
 
@@ -542,6 +558,8 @@ mod tests {
             toolkit: "gmail".to_string(),
             connection_id: None,
             usage: ComposioUsageHandle::default(),
+            max_items: None,
+            sync_depth_days: None,
         };
         let res = ctx.execute("GMAIL_FETCH_EMAILS", None).await;
         // The actual HTTP call will fail in the unit-test sandbox, but
@@ -575,6 +593,8 @@ mod tests {
             toolkit: "gmail".to_string(),
             connection_id: None,
             usage: ComposioUsageHandle::default(),
+            max_items: None,
+            sync_depth_days: None,
         };
         let res = ctx.execute("GMAIL_FETCH_EMAILS", None).await;
         let err = res.expect_err("no backend session must error");

src/openhuman/task_sources/ops.rs

@@ -153,6 +153,8 @@ pub async fn preview_filter(
         toolkit: provider.as_str().to_string(),
         connection_id: connection_id.filter(|s| !s.trim().is_empty()),
         usage: Default::default(),
+        max_items: None,
+        sync_depth_days: None,
     };
     let max = max.unwrap_or(config.task_sources.max_tasks_per_fetch);
     let fetch_filter = filter::to_fetch_filter(&filter_spec, max);
@@ -179,6 +181,8 @@ pub async fn list_databases(
         toolkit: provider.as_str().to_string(),
         connection_id: connection_id.filter(|s| !s.trim().is_empty()),
         usage: Default::default(),
+        max_items: None,
+        sync_depth_days: None,
     };
     let databases = provider_impl
         .list_databases(&ctx)

src/openhuman/task_sources/pipeline.rs

@@ -102,6 +102,8 @@ async fn run_inner(
         toolkit: source.provider.as_str().to_string(),
         connection_id: source.connection_id.clone(),
         usage: Default::default(),
+        max_items: None,
+        sync_depth_days: None,
     };
 
     let fetch_filter = filter::to_fetch_filter(&source.filter, source.max_tasks_per_fetch);