refactor(memory_sync): internalize the cap math into the orchestrator (single source of truth) (#3937)

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

@@ -184,6 +184,7 @@ impl ComposioProvider for GmailProvider {
     }
 }
 
-// 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.
+// The `max_items` cap math (`ItemCap`) lives in the orchestrator now that it is
+// the sole consumer; the `sync_depth_days` date floor (`epoch_floor_from_depth`)
+// stays in `super::super::helpers` because `gmail::source` builds an
+// `after:<epoch>` filter from it.

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

@@ -46,21 +46,13 @@ 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
-}
+// ── Window helper ────────────────────────────────────────────────────────
+//
+// The per-sync `max_items` cap math (`ItemCap` + `pages_for_max_items`) lives
+// in the orchestrator now that it is the sole consumer — see
+// [`super::orchestrator`]. `epoch_floor_from_depth` stays here because it is a
+// provider-facing window helper (gmail/source.rs builds an `after:<epoch>`
+// filter from it), not orchestrator-internal cap math.
 
 /// 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.
@@ -70,86 +62,10 @@ pub(crate) fn epoch_floor_from_depth(sync_depth_days: u32) -> 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 {
+mod window_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);
@@ -161,61 +77,6 @@ mod cap_helper_tests {
             "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

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

@@ -8,11 +8,13 @@
 //! [`SyncOutcome`]. The copy-paste is exactly how the page-granular cap bug
 //! (#3304) ended up in five providers and was missed in a sixth.
 //!
-//! [`ItemCap`] (PR #3304) centralised the cap *math*; this module centralises
-//! the *control flow*. A provider now implements only the slim
-//! [`IncrementalSource`] primitives and rides [`run_sync`], inheriting the
-//! `max_items` cap, the `sync_depth_days` window, dedup, cursor advance, and
-//! budget enforcement for free.
+//! [`ItemCap`] (PR #3304) centralised the cap *math*; this module owns it and
+//! the *control flow*. Now that every provider rides [`run_sync`], `ItemCap`
+//! lives here (orchestrator-private), not in the shared `helpers` module — the
+//! cap rule exists in exactly one place. A provider implements only the slim
+//! [`IncrementalSource`] primitives and inherits the `max_items` cap, the
+//! `sync_depth_days` window, dedup, cursor advance, and budget enforcement for
+//! free.
 //!
 //! ## Scopes: flat AND nested in one loop
 //!
@@ -39,10 +41,86 @@
 use async_trait::async_trait;
 use serde_json::{json, Value};
 
-use super::helpers::ItemCap;
 use super::sync_state::SyncState;
 use super::{ProviderContext, SyncOutcome, SyncReason};
 
+/// 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".
+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
+}
+
+/// 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 (#3304) ended up in five providers and was missed
+/// in a sixth. Now that every provider rides [`run_sync`], this type and its
+/// page math are **orchestrator-private**: the cap rule lives in exactly one
+/// place — construct it from `ctx.max_items`, derive the page cap, clamp each
+/// batch before ingest, 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)]
+struct ItemCap {
+    cap: Option<usize>,
+    persisted: usize,
+}
+
+impl ItemCap {
+    /// Build from a source's `max_items` value (`None` = uncapped).
+    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.
+    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.
+    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.
+    fn is_reached(&self) -> bool {
+        matches!(self.remaining(), Some(0))
+    }
+
+    /// Record `n` newly-persisted items against the budget.
+    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.
+    fn clamp_batch<T>(&self, batch: &mut Vec<T>) {
+        if let Some(remaining) = self.remaining() {
+            if batch.len() > remaining {
+                batch.truncate(remaining);
+            }
+        }
+    }
+}
+
 /// A unit of work to iterate within one sync pass.
 ///
 /// Flat providers return a single [`SyncScope::flat`]; nested providers return
@@ -1391,4 +1469,75 @@ mod tests {
             .expect_err("without tolerance a scope fetch error propagates");
         assert!(err.contains("scope fetch failure"), "got: {err}");
     }
+
+    // ── ItemCap / pages_for_max_items (relocated from helpers.rs, #3902) ──
+
+    #[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 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");
+    }
 }