From 7758148c2ce53d2571e65298595297251fd2c2d1 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Mon, 22 Jun 2026 01:55:22 +0530 Subject: [PATCH] refactor(memory_sync): generic Composio sync orchestrator + convert Notion (Refs #3333) (#3852) Co-authored-by: Claude --- .../memory_sync/composio/providers/mod.rs | 1 + .../composio/providers/notion/mod.rs | 1 + .../composio/providers/notion/provider.rs | 764 +-------------- .../composio/providers/notion/source.rs | 550 +++++++++++ .../composio/providers/orchestrator.rs | 927 ++++++++++++++++++ .../memory_sync_providers_raw_coverage_e2e.rs | 81 ++ 6 files changed, 1570 insertions(+), 754 deletions(-) create mode 100644 src/openhuman/memory_sync/composio/providers/notion/source.rs create mode 100644 src/openhuman/memory_sync/composio/providers/orchestrator.rs diff --git a/src/openhuman/memory_sync/composio/providers/mod.rs b/src/openhuman/memory_sync/composio/providers/mod.rs index 640b014a2..8236cd6df 100644 --- a/src/openhuman/memory_sync/composio/providers/mod.rs +++ b/src/openhuman/memory_sync/composio/providers/mod.rs @@ -34,6 +34,7 @@ mod descriptions; pub(crate) mod helpers; +pub(crate) mod orchestrator; mod scope_lookup; pub mod tool_scope; mod traits; diff --git a/src/openhuman/memory_sync/composio/providers/notion/mod.rs b/src/openhuman/memory_sync/composio/providers/notion/mod.rs index 44bc4f4b9..228a853d2 100644 --- a/src/openhuman/memory_sync/composio/providers/notion/mod.rs +++ b/src/openhuman/memory_sync/composio/providers/notion/mod.rs @@ -1,5 +1,6 @@ mod ingest; mod provider; +mod source; mod sync; #[cfg(test)] mod tests; diff --git a/src/openhuman/memory_sync/composio/providers/notion/provider.rs b/src/openhuman/memory_sync/composio/providers/notion/provider.rs index 546b0d980..490a657bd 100644 --- a/src/openhuman/memory_sync/composio/providers/notion/provider.rs +++ b/src/openhuman/memory_sync/composio/providers/notion/provider.rs @@ -18,64 +18,31 @@ use async_trait::async_trait; use serde_json::{json, Value}; -use super::ingest::ingest_page_into_memory_tree; +use super::source::run_notion_sync; use super::sync; -use crate::openhuman::config::Config; -use crate::openhuman::memory_sync::composio::providers::sync_state::{extract_item_id, SyncState}; +use crate::openhuman::memory_sync::composio::providers::sync_state::extract_item_id; use crate::openhuman::memory_sync::composio::providers::{ first_array_str, merge_extra, pick_str, resolve_sync_interval_secs, ComposioProvider, CuratedTool, NormalizedTask, ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason, TaskContainer, TaskFetchFilter, TaskKind, }; -use futures::StreamExt; pub(crate) const ACTION_GET_ABOUT_ME: &str = "NOTION_GET_ABOUT_ME"; pub(crate) const ACTION_FETCH_DATA: &str = "NOTION_FETCH_DATA"; -/// Per-page action that returns the page's rendered **body** as markdown -/// (paragraphs, headings, lists, body tables). `FETCH_DATA` only returns -/// metadata + properties; this fills in the actual document content for -/// free-form pages. One request per page (budget-counted). -pub(crate) const ACTION_GET_PAGE_MARKDOWN: &str = "NOTION_GET_PAGE_MARKDOWN"; pub(crate) const ACTION_QUERY_DATABASE: &str = "NOTION_QUERY_DATABASE"; pub(crate) const ACTION_SEARCH_NOTION_PAGE: &str = "NOTION_SEARCH_NOTION_PAGE"; -/// Page size per API call. -const PAGE_SIZE: u32 = 25; - -/// Larger page size for initial sync after OAuth. -const INITIAL_PAGE_SIZE: u32 = 50; - -/// Maximum pages per sync pass. -const MAX_PAGES_PER_SYNC: u32 = 20; - /// Paths for extracting a page's unique ID. -const PAGE_ID_PATHS: &[&str] = &["id", "data.id", "pageId", "data.pageId"]; +pub(crate) const PAGE_ID_PATHS: &[&str] = &["id", "data.id", "pageId", "data.pageId"]; /// Paths for extracting the `last_edited_time` used as sync cursor. -const PAGE_EDITED_PATHS: &[&str] = &[ +pub(crate) const PAGE_EDITED_PATHS: &[&str] = &[ "last_edited_time", "data.last_edited_time", "lastEditedTime", "data.lastEditedTime", ]; -/// Max in-flight ingests per page. DB writes serialize anyway and the -/// cloud embedder has rate limits, so keep this small. -const INGEST_CONCURRENCY: usize = 8; - -/// Max in-flight `GET_PAGE_MARKDOWN` body fetches per page. Kept small to -/// respect Notion/Composio rate limits while still overlapping the per-request -/// round-trips instead of paying them strictly serially. -const BODY_FETCH_CONCURRENCY: usize = 5; - -/// How many pending pages we may fetch bodies for, given the remaining daily -/// request budget. Pure so it can be unit-tested: it reproduces the old serial -/// "check `budget_exhausted()` before each fetch, break when it hits zero" -/// behaviour as a single up-front clamp (one body fetch == one request). -fn body_fetch_count(pending_len: usize, budget_remaining: u32) -> usize { - pending_len.min(budget_remaining as usize) -} - pub struct NotionProvider; impl NotionProvider { @@ -161,376 +128,13 @@ impl ComposioProvider for NotionProvider { }) } + /// Incremental sync. Notion was the first provider migrated to the generic + /// [`orchestrator`](crate::openhuman::memory_sync::composio::providers::orchestrator): + /// the per-item loop, dedup, `max_items` cap, `sync_depth_days` window, and + /// cursor handling all live in `run_sync`; the Notion-specific primitives + /// (page fetch, dedup key, body fetch, ingest) live in [`super::source`]. async fn sync(&self, ctx: &ProviderContext, reason: SyncReason) -> Result { - let started_at_ms = sync::now_ms(); - let connection_id = ctx - .connection_id - .clone() - .unwrap_or_else(|| "default".to_string()); - - tracing::info!( - connection_id = %connection_id, - reason = reason.as_str(), - "[composio:notion] incremental sync starting" - ); - - // ── Step 1: load persistent sync state ────────────────────── - let Some(memory) = ctx.memory_client() else { - return Err("[composio:notion] memory client not ready".to_string()); - }; - let mut state = SyncState::load(&memory, "notion", &connection_id).await?; - - // ── Step 2: check daily budget ────────────────────────────── - if state.budget_exhausted() { - tracing::info!( - connection_id = %connection_id, - "[composio:notion] daily request budget exhausted, skipping sync" - ); - return Ok(SyncOutcome { - toolkit: "notion".to_string(), - connection_id: Some(connection_id), - reason: reason.as_str().to_string(), - items_ingested: 0, - started_at_ms, - finished_at_ms: sync::now_ms(), - summary: "notion sync skipped: daily budget exhausted".to_string(), - details: json!({ "budget_exhausted": true }), - }); - } - - // ── Step 3: paginated incremental fetch ───────────────────── - let page_size = match reason { - SyncReason::ConnectionCreated => INITIAL_PAGE_SIZE, - _ => 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 = 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 = None; - let mut notion_cursor: Option = None; - // Track whether any per-item ingest failed this pass. If so, we hold - // the persistent cursor — `last_edited_time > {cursor}` on the next - // sync would otherwise exclude the failed item, and because the new - // memory-tree pipeline (#2885) is delete-first, an *edited* page that - // failed to re-ingest is left with neither old nor new chunks 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; - - for page_num in 0..effective_max_pages { - if state.budget_exhausted() { - tracing::info!( - page = page_num, - "[composio:notion] budget exhausted mid-sync, stopping pagination" - ); - break; - } - - let mut args = json!({ - "page_size": page_size, - "filter": { "value": "page", "property": "object" }, - "sort": { "direction": "descending", "timestamp": "last_edited_time" } - }); - if let Some(ref cursor) = notion_cursor { - args["start_cursor"] = json!(cursor); - } - - let resp = ctx - .execute(ACTION_FETCH_DATA, Some(args)) - .await - .map_err(|e| { - format!("[composio:notion] {ACTION_FETCH_DATA} page {page_num}: {e:#}") - })?; - - state.record_requests(1); - - if !resp.successful { - let err = resp - .error - .clone() - .unwrap_or_else(|| "provider reported failure".to_string()); - let _ = state.save(&memory).await; - return Err(format!( - "[composio:notion] {ACTION_FETCH_DATA} page {page_num}: {err}" - )); - } - - let results = sync::extract_results(&resp.data); - total_fetched += results.len(); - - if results.is_empty() { - tracing::debug!( - page = page_num, - "[composio:notion] empty page, stopping pagination" - ); - break; - } - - // ── Step 4a: dedupe + decide which pages to ingest ────── - 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 4a.5: fetch each pending page's BODY markdown (bounded concurrency) ── - // FETCH_DATA only returned metadata + properties. Pull the - // rendered page body (paragraphs, lists, body tables) per page so - // free-form documents ingest as real content (multi-chunk) rather - // than a single metadata chunk. One request per page — budget - // counted. Runs AFTER the depth-floor + max-items cap so we only - // fetch bodies for pages we'll actually ingest. On budget - // exhaustion or error we leave `markdown_body` None and ingest - // falls back to the metadata-only body. - // - // We pre-clamp the number of bodies to the remaining daily request - // budget (reproducing the old serial "check before each fetch, - // break at zero" semantics as a single up-front decision) and then - // fire the fetches `BODY_FETCH_CONCURRENCY`-at-a-time. `buffered` - // preserves input order, so result[i] maps back to pending[i]. - let fetch_count = body_fetch_count(pending.len(), state.budget_remaining()); - if fetch_count < pending.len() { - tracing::info!( - page = page_num, - fetch_count, - pending = pending.len(), - "[composio:notion] daily budget caps body fetch — \ - remaining pages ingest metadata-only" - ); - } - if fetch_count > 0 { - // Snapshot the page ids so the fetch futures don't borrow - // `pending` (we write results back into it afterwards). - let page_ids: Vec = pending[..fetch_count] - .iter() - .map(|p| p.page_id.clone()) - .collect(); - - // Materialize the futures into a Vec before `buffered` so the - // higher-ranked closure lifetime stays tied to this stack frame - // (avoids "implementation of FnOnce is not general enough"). - let body_futs: Vec<_> = page_ids - .into_iter() - .enumerate() - .map(|(idx, page_id)| { - let ctx = &ctx; - async move { - match ctx - .execute( - ACTION_GET_PAGE_MARKDOWN, - Some(json!({ "page_id": &page_id })), - ) - .await - { - Ok(resp) if resp.successful => { - let markdown = sync::extract_page_markdown(&resp.data); - // Empirical visibility (INFO so it shows at - // default log level): on the first body - // fetch, log the markdown length + raw - // envelope keys to confirm the response - // field / arg name on a live run. - if page_num == 0 && idx == 0 { - tracing::info!( - page_id = %page_id, - markdown_chars = - markdown.as_ref().map(|s| s.len()).unwrap_or(0), - raw_keys = ?resp - .data - .as_object() - .map(|o| o.keys().cloned().collect::>()), - "[composio:notion] GET_PAGE_MARKDOWN sample (empirical check)" - ); - } - if markdown.is_none() { - tracing::warn!( - page_id = %page_id, - "[composio:notion] GET_PAGE_MARKDOWN returned no \ - markdown field — metadata-only fallback" - ); - } - markdown - } - Ok(resp) => { - tracing::warn!( - page_id = %page_id, - error = ?resp.error, - "[composio:notion] GET_PAGE_MARKDOWN failed — \ - metadata-only fallback" - ); - None - } - Err(e) => { - tracing::warn!( - page_id = %page_id, - error = %e, - "[composio:notion] GET_PAGE_MARKDOWN execute error — \ - metadata-only fallback" - ); - None - } - } - } - }) - .collect(); - - let bodies: Vec> = futures::stream::iter(body_futs) - .buffered(BODY_FETCH_CONCURRENCY) - .collect() - .await; - - // Count every request we fired (success or failure), matching - // the old per-call `record_requests(1)`. - state.record_requests(fetch_count as u32); - - for (p, body) in pending.iter_mut().zip(bodies.into_iter()) { - p.markdown_body = body; - } - } - - // ── Step 4b: ingest queued pages (bounded concurrency) ── - let ingestor = MemoryTreeIngestor { - config: ctx.config.as_ref(), - connection_id: &connection_id, - }; - let outcome = ingest_pending_buffered(&ingestor, pending, INGEST_CONCURRENCY).await; - for key in &outcome.synced_keys { - 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, - "[composio:notion] reached cursor boundary, stopping" - ); - 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() { - tracing::debug!(page = page_num, "[composio:notion] no next cursor, done"); - break; - } - } - - // ── Step 5: advance cursor and save state ─────────────────── - // - // 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). - // 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, - 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?; - - let finished_at_ms = sync::now_ms(); - let summary = format!( - "notion sync ({reason}): fetched {total_fetched}, persisted {total_persisted} new, \ - budget remaining {remaining}", - reason = reason.as_str(), - remaining = state.budget_remaining(), - ); - tracing::info!( - connection_id = %connection_id, - elapsed_ms = finished_at_ms.saturating_sub(started_at_ms), - total_fetched, - total_persisted, - budget_remaining = state.budget_remaining(), - "[composio:notion] incremental sync complete" - ); - - Ok(SyncOutcome { - toolkit: "notion".to_string(), - connection_id: Some(connection_id), - reason: reason.as_str().to_string(), - items_ingested: total_persisted, - started_at_ms, - finished_at_ms, - summary, - details: json!({ - "results_fetched": total_fetched, - "results_persisted": total_persisted, - "budget_remaining": state.budget_remaining(), - "cursor": state.cursor, - "synced_ids_total": state.synced_ids.len(), - }), - }) + run_notion_sync(ctx, reason).await } async fn fetch_tasks( @@ -800,351 +404,3 @@ fn extract_database_title(db: &serde_json::Value) -> Option { let text = text.trim(); (!text.is_empty()).then(|| text.to_string()) } - -/// One page that passed dedupe in pass 1 and is queued for concurrent -/// ingest in pass 2. Borrows the raw page `Value` out of the current -/// page's `results` (same scope — no clone needed). -struct PendingIngest<'a> { - sync_key: String, - page_id: String, - title: String, - edited_time: Option, - page: &'a Value, - /// Rendered page body (markdown) fetched per-page via - /// `NOTION_GET_PAGE_MARKDOWN` after dedupe, before ingest. `None` when the - /// body fetch was skipped (budget) or failed — ingest falls back to the - /// metadata-only body. - markdown_body: Option, -} - -/// Folded result of [`ingest_pending_buffered`]. Every field is -/// order-independent, so the concurrent stage can accumulate into it -/// regardless of the order ingests complete. -#[derive(Default)] -struct BufferedIngestOutcome { - /// `sync_key`s whose ingest succeeded — the caller marks each synced. - synced_keys: Vec, - /// Number of pages persisted (equals `synced_keys.len()`). - persisted: usize, - /// Whether any per-item ingest failed (the caller holds the cursor). - had_failures: bool, -} - -/// Seam over "ingest one Notion page", so the bounded-concurrency driver -/// can be unit-tested with a fake that records peak in-flight calls -/// without a real memory tree or embedder. -#[async_trait] -trait PageIngestor { - async fn ingest( - &self, - page_id: &str, - title: &str, - edited_time: Option<&str>, - page: &Value, - markdown_body: Option<&str>, - ) -> anyhow::Result; -} - -/// Production ingestor: routes into the memory-tree pipeline (#2885) via -/// [`ingest_page_into_memory_tree`]. -struct MemoryTreeIngestor<'c> { - config: &'c Config, - connection_id: &'c str, -} - -#[async_trait] -impl PageIngestor for MemoryTreeIngestor<'_> { - async fn ingest( - &self, - page_id: &str, - title: &str, - edited_time: Option<&str>, - page: &Value, - markdown_body: Option<&str>, - ) -> anyhow::Result { - ingest_page_into_memory_tree( - self.config, - self.connection_id, - page_id, - title, - edited_time, - page, - markdown_body, - ) - .await - } -} - -/// Pass 1 (pure, no I/O): scan one page of `results`, advance -/// `newest_edited_time`, skip already-synced items, and collect the -/// pages still needing ingest. Returns the queued items and whether we -/// crossed the persistent cursor boundary (the signal to stop -/// paginating). All order-dependent decisions (cursor/timestamp) live -/// here — never in the concurrent stage. -fn select_pending<'a>( - results: &'a [Value], - state: &SyncState, - newest_edited_time: &mut Option, -) -> (Vec>, bool) { - let mut hit_cursor_boundary = false; - let mut pending: Vec = Vec::new(); - for page in results { - let Some(page_id) = extract_item_id(page, PAGE_ID_PATHS) else { - tracing::debug!("[composio:notion] page missing ID, skipping"); - continue; - }; - - let edited_time = extract_item_id(page, PAGE_EDITED_PATHS); - - // Track the newest edited time for cursor advancement. - if let Some(ref et) = edited_time { - if newest_edited_time - .as_ref() - .is_none_or(|existing| et > existing) - { - *newest_edited_time = Some(et.clone()); - } - } - - let sync_key = match &edited_time { - Some(et) => format!("{page_id}@{et}"), - None => page_id.clone(), - }; - - // Older than cursor AND already synced → caught up. - if let (Some(ref cursor), Some(ref et)) = (&state.cursor, &edited_time) { - if et <= cursor && state.is_synced(&sync_key) { - hit_cursor_boundary = true; - continue; - } - } - - if state.is_synced(&sync_key) { - continue; - } - - let title_text = - sync::extract_page_title(page).unwrap_or_else(|| format!("Notion page {page_id}")); - let title = format!("Notion: {title_text}"); - - pending.push(PendingIngest { - sync_key, - page_id, - title, - edited_time, - page, - markdown_body: None, - }); - } - (pending, hit_cursor_boundary) -} - -/// Pass 2: ingest the queued pages with bounded concurrency. Overlaps -/// the per-item embedding RTT (`buffer_unordered`, up to `concurrency` -/// in flight) and folds results into an order-independent -/// [`BufferedIngestOutcome`]. Unordered is correct here (unlike the -/// order-aligned GitHub reader): nothing downstream depends on -/// completion order — successes are keyed by `sync_key`. -async fn ingest_pending_buffered( - ingestor: &I, - pending: Vec>, - concurrency: usize, -) -> BufferedIngestOutcome { - // Materialize the per-item futures into a Vec before `buffer_unordered` - // so the spawned sync future keeps concrete lifetimes / `Send`. - let ingest_futs = pending - .into_iter() - .map(|p| async move { - let res = ingestor - .ingest( - &p.page_id, - &p.title, - p.edited_time.as_deref(), - p.page, - p.markdown_body.as_deref(), - ) - .await; - (p.sync_key, p.page_id, res) - }) - .collect::>(); - - let mut outcome = BufferedIngestOutcome::default(); - let mut ingest_stream = futures::stream::iter(ingest_futs).buffer_unordered(concurrency); - while let Some((sync_key, page_id, res)) = ingest_stream.next().await { - match res { - Ok(_chunks_written) => { - outcome.synced_keys.push(sync_key); - outcome.persisted += 1; - } - Err(e) => { - outcome.had_failures = true; - tracing::warn!( - page_id = %page_id, - error = %e, - "[composio:notion] failed to ingest page into memory_tree (continuing)" - ); - } - } - } - outcome -} - -#[cfg(test)] -mod body_fetch_count_tests { - use super::body_fetch_count; - - #[test] - fn clamps_to_remaining_budget_when_budget_is_the_limit() { - // 10 pending pages but only 3 requests left today → fetch 3. - assert_eq!(body_fetch_count(10, 3), 3); - } - - #[test] - fn fetches_all_when_budget_exceeds_pending() { - assert_eq!(body_fetch_count(4, 100), 4); - } - - #[test] - fn zero_budget_fetches_nothing() { - // Mirrors the old serial "budget_exhausted() before first fetch" break. - assert_eq!(body_fetch_count(7, 0), 0); - } - - #[test] - fn zero_pending_fetches_nothing() { - assert_eq!(body_fetch_count(0, 50), 0); - } -} - -#[cfg(test)] -mod buffered_tests { - use super::*; - use serde_json::json; - use std::sync::atomic::{AtomicUsize, Ordering}; - use std::sync::Arc; - - /// Fake ingestor: records the peak number of concurrent in-flight - /// `ingest` calls and can be told to fail one specific `page_id`. No - /// memory tree or embedder involved — lets us assert the concurrency - /// bound and overlap deterministically. - struct CountingIngestor { - in_flight: AtomicUsize, - peak: AtomicUsize, - fail_page: Option, - } - - impl CountingIngestor { - fn new(fail_page: Option<&str>) -> Arc { - Arc::new(Self { - in_flight: AtomicUsize::new(0), - peak: AtomicUsize::new(0), - fail_page: fail_page.map(str::to_string), - }) - } - } - - #[async_trait] - impl PageIngestor for CountingIngestor { - async fn ingest( - &self, - page_id: &str, - _title: &str, - _edited_time: Option<&str>, - _page: &Value, - _markdown_body: Option<&str>, - ) -> anyhow::Result { - let now = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1; - self.peak.fetch_max(now, Ordering::SeqCst); - // Yield a few times so futures genuinely interleave and the - // peak counter reflects real overlap, not accidental serial run. - for _ in 0..4 { - tokio::task::yield_now().await; - } - self.in_flight.fetch_sub(1, Ordering::SeqCst); - if self.fail_page.as_deref() == Some(page_id) { - Err(anyhow::anyhow!("forced failure for {page_id}")) - } else { - Ok(1) - } - } - } - - fn make_pages(n: usize) -> Vec { - (0..n).map(|i| json!({ "id": format!("p{i}") })).collect() - } - - fn make_pending<'a>(pages: &'a [Value]) -> Vec> { - pages - .iter() - .enumerate() - .map(|(i, page)| PendingIngest { - sync_key: format!("k{i}"), - page_id: format!("p{i}"), - title: format!("Notion: page {i}"), - edited_time: None, - page, - markdown_body: None, - }) - .collect() - } - - #[tokio::test] - async fn ingest_pending_buffered_bounds_and_overlaps() { - let ingestor = CountingIngestor::new(None); - let pages = make_pages(20); - let pending = make_pending(&pages); - - let outcome = ingest_pending_buffered(ingestor.as_ref(), pending, 8).await; - - assert_eq!(outcome.persisted, 20, "all pages persisted"); - assert_eq!(outcome.synced_keys.len(), 20); - assert!(!outcome.had_failures); - - let peak = ingestor.peak.load(Ordering::SeqCst); - assert!(peak <= 8, "peak in-flight {peak} exceeded the bound of 8"); - assert!(peak >= 2, "peak in-flight {peak} shows no real overlap"); - } - - #[tokio::test] - async fn ingest_pending_buffered_skips_failures_order_independent() { - let ingestor = CountingIngestor::new(Some("p2")); - let pages = make_pages(5); - let pending = make_pending(&pages); - - let outcome = ingest_pending_buffered(ingestor.as_ref(), pending, 4).await; - - assert_eq!(outcome.persisted, 4, "the one failed ingest is not counted"); - assert!(outcome.had_failures); - assert_eq!(outcome.synced_keys.len(), 4); - assert!( - !outcome.synced_keys.contains(&"k2".to_string()), - "the failed page's sync_key must not be marked synced" - ); - } - - #[test] - fn select_pending_tracks_newest_skips_synced_and_detects_boundary() { - let mut state = SyncState::new("notion", "conn1"); - state.cursor = Some("2026-04-15T00:00:00Z".to_string()); - // Page B is already synced and older than the cursor. - state.mark_synced("b@2026-04-01T00:00:00Z"); - - let pages = vec![ - json!({ "id": "a", "last_edited_time": "2026-05-01T00:00:00Z" }), - json!({ "id": "b", "last_edited_time": "2026-04-01T00:00:00Z" }), - json!({ "last_edited_time": "2026-03-01T00:00:00Z" }), // no id → skipped - ]; - - let mut newest: Option = None; - let (pending, hit_boundary) = select_pending(&pages, &state, &mut newest); - - assert_eq!(pending.len(), 1, "only the new page A is queued"); - assert_eq!(pending[0].page_id, "a"); - assert_eq!(pending[0].sync_key, "a@2026-05-01T00:00:00Z"); - assert!( - hit_boundary, - "older synced page B trips the cursor boundary" - ); - assert_eq!(newest.as_deref(), Some("2026-05-01T00:00:00Z")); - } -} diff --git a/src/openhuman/memory_sync/composio/providers/notion/source.rs b/src/openhuman/memory_sync/composio/providers/notion/source.rs new file mode 100644 index 000000000..77626533f --- /dev/null +++ b/src/openhuman/memory_sync/composio/providers/notion/source.rs @@ -0,0 +1,550 @@ +//! Notion's [`IncrementalSource`] primitives. +//! +//! Notion is the first provider converted to the generic +//! [`crate::openhuman::memory_sync::composio::providers::orchestrator`]: +//! [`NotionProvider::sync`](super::provider::NotionProvider) delegates to +//! [`run_notion_sync`], which runs [`NotionSource`] through +//! [`orchestrator::run_sync`]. The orchestrator owns the control flow (budget, +//! pagination bound, dedup, the `sync_depth_days` window, the precise +//! `max_items` clamp, cursor advance/hold, state persistence); this module +//! supplies only the Notion-specific shapes. +//! +//! Notion is a **flat** provider — [`NotionSource::preamble`] returns a single +//! [`SyncScope::flat`] and the orchestrator pages straight through +//! `NOTION_FETCH_DATA`. Per-item dedup is keyed by `{page_id}@{edited_time}` so +//! an *edited* page (its `last_edited_time` advances) is re-ingested, exactly as +//! before. The rendered page **body** is fetched per page via +//! `NOTION_GET_PAGE_MARKDOWN` inside [`NotionSource::ingest`] (budget-counted), +//! then handed to the memory-tree pipeline. + +use async_trait::async_trait; +use futures::StreamExt; +use serde_json::{json, Value}; + +use super::ingest::ingest_page_into_memory_tree; +use super::provider::{ACTION_FETCH_DATA, PAGE_EDITED_PATHS, PAGE_ID_PATHS}; +use super::sync; +use crate::openhuman::config::Config; +use crate::openhuman::memory_sync::composio::providers::orchestrator::{ + self, IncrementalSource, IngestOutcome, PageFetch, SyncItem, SyncScope, +}; +use crate::openhuman::memory_sync::composio::providers::sync_state::{extract_item_id, SyncState}; +use crate::openhuman::memory_sync::composio::providers::{ + ProviderContext, SyncOutcome, SyncReason, +}; + +/// Per-page action that returns the page's rendered **body** as markdown +/// (paragraphs, headings, lists, body tables). `NOTION_FETCH_DATA` only returns +/// metadata + properties; this fills in the actual document content for +/// free-form pages. One request per page (budget-counted). +pub(crate) const ACTION_GET_PAGE_MARKDOWN: &str = "NOTION_GET_PAGE_MARKDOWN"; + +/// Page size per API call. +const PAGE_SIZE: u32 = 25; + +/// Larger page size for initial sync after OAuth. +const INITIAL_PAGE_SIZE: u32 = 50; + +/// Maximum pages per sync pass. +const MAX_PAGES_PER_SYNC: u32 = 20; + +/// Max in-flight ingests per page. DB writes serialize anyway and the +/// cloud embedder has rate limits, so keep this small. +const INGEST_CONCURRENCY: usize = 8; + +/// Max in-flight `GET_PAGE_MARKDOWN` body fetches per page. Kept small to +/// respect Notion/Composio rate limits while still overlapping the per-request +/// round-trips instead of paying them strictly serially. +const BODY_FETCH_CONCURRENCY: usize = 5; + +/// How many pending pages we may fetch bodies for, given the remaining daily +/// request budget. Pure so it can be unit-tested: it reproduces the old serial +/// "check `budget_exhausted()` before each fetch, break when it hits zero" +/// behaviour as a single up-front clamp (one body fetch == one request). +fn body_fetch_count(pending_len: usize, budget_remaining: u32) -> usize { + pending_len.min(budget_remaining as usize) +} + +/// Notion's [`IncrementalSource`] — supplies only the provider-specific shapes; +/// the orchestrator owns everything else. +pub(crate) struct NotionSource; + +/// Entry point used by [`super::provider::NotionProvider::sync`]. +pub(crate) async fn run_notion_sync( + ctx: &ProviderContext, + reason: SyncReason, +) -> Result { + orchestrator::run_sync(&NotionSource, ctx, reason).await +} + +#[async_trait] +impl IncrementalSource for NotionSource { + fn toolkit(&self) -> &'static str { + "notion" + } + + fn page_size(&self, reason: SyncReason) -> u32 { + match reason { + SyncReason::ConnectionCreated => INITIAL_PAGE_SIZE, + _ => PAGE_SIZE, + } + } + + fn max_pages(&self) -> u32 { + MAX_PAGES_PER_SYNC + } + + /// Notion is flat — one implicit scope, no identity resolution needed here + /// (the user profile is fetched separately by `on_connection_created`). + async fn preamble( + &self, + _ctx: &ProviderContext, + _state: &mut SyncState, + ) -> Result, String> { + Ok(vec![SyncScope::flat()]) + } + + async fn fetch_page( + &self, + ctx: &ProviderContext, + _scope: &SyncScope, + cursor: Option<&str>, + reason: SyncReason, + state: &mut SyncState, + ) -> Result { + let mut args = json!({ + "page_size": self.page_size(reason), + "filter": { "value": "page", "property": "object" }, + "sort": { "direction": "descending", "timestamp": "last_edited_time" } + }); + if let Some(cursor) = cursor { + args["start_cursor"] = json!(cursor); + } + + // A transport error never reached Composio — `?` returns before we + // record. A completed round-trip is recorded against the daily budget + // *before* we surface a provider-reported failure, so a broken + // connection cannot make unlimited billable failed page calls. + let resp = ctx + .execute(ACTION_FETCH_DATA, Some(args)) + .await + .map_err(|e| format!("[composio:notion] {ACTION_FETCH_DATA}: {e:#}"))?; + state.record_requests(1); + + if !resp.successful { + let err = resp + .error + .clone() + .unwrap_or_else(|| "provider reported failure".to_string()); + return Err(format!("[composio:notion] {ACTION_FETCH_DATA}: {err}")); + } + + Ok(PageFetch { + items: sync::extract_results(&resp.data), + next: sync::extract_notion_cursor(&resp.data), + }) + } + + /// Dedup key is `{page_id}@{last_edited_time}` (or the bare id when no + /// edited time) so an edited page re-ingests while an unchanged one is + /// skipped. + fn item_dedup_key(&self, item: &Value) -> Option { + let page_id = extract_item_id(item, PAGE_ID_PATHS)?; + match extract_item_id(item, PAGE_EDITED_PATHS) { + Some(edited) => Some(format!("{page_id}@{edited}")), + None => Some(page_id), + } + } + + fn item_sort_ts(&self, item: &Value) -> Option { + extract_item_id(item, PAGE_EDITED_PATHS) + } + + async fn ingest( + &self, + ctx: &ProviderContext, + _scope: &SyncScope, + state: &mut SyncState, + items: Vec, + ) -> IngestOutcome { + let connection_id = ctx.connection_id.as_deref().unwrap_or("default"); + + // Build the per-page ingest queue, re-extracting id/title from each raw + // page. `sync_key`/`edited_time` come straight from the orchestrator's + // dedup/sort decision so they stay consistent with `mark_synced`. + let mut pending: Vec = items + .into_iter() + .filter_map(|it| { + let page_id = extract_item_id(&it.raw, PAGE_ID_PATHS)?; + let title_text = sync::extract_page_title(&it.raw) + .unwrap_or_else(|| format!("Notion page {page_id}")); + Some(PendingIngest { + sync_key: it.dedup_key, + page_id, + title: format!("Notion: {title_text}"), + edited_time: it.sort_ts, + page: it.raw, + markdown_body: None, + }) + }) + .collect(); + + // ── Per-page BODY markdown fetch (bounded concurrency) ────────── + // `NOTION_FETCH_DATA` returned metadata + properties only. Pull the + // rendered page body per page so free-form documents ingest as real + // multi-chunk content. One request per page — budget counted. We + // pre-clamp the number of bodies to the remaining daily budget + // (reproducing the old serial "check before each fetch, break at zero" + // semantics as a single up-front decision) and fire the fetches + // `BODY_FETCH_CONCURRENCY`-at-a-time; `buffered` preserves input order + // so result[i] maps back to pending[i]. + let fetch_count = body_fetch_count(pending.len(), state.budget_remaining()); + if fetch_count < pending.len() { + tracing::info!( + fetch_count, + pending = pending.len(), + "[composio:notion] daily budget caps body fetch — \ + remaining pages ingest metadata-only" + ); + } + if fetch_count > 0 { + // Snapshot the page ids so the fetch futures don't borrow `pending` + // (we write results back into it afterwards). + let page_ids: Vec = pending[..fetch_count] + .iter() + .map(|p| p.page_id.clone()) + .collect(); + + let body_futs: Vec<_> = page_ids + .into_iter() + .map(|page_id| { + let ctx = &ctx; + async move { + match ctx + .execute( + ACTION_GET_PAGE_MARKDOWN, + Some(json!({ "page_id": &page_id })), + ) + .await + { + Ok(resp) if resp.successful => { + let markdown = sync::extract_page_markdown(&resp.data); + if markdown.is_none() { + tracing::warn!( + page_id = %page_id, + "[composio:notion] GET_PAGE_MARKDOWN returned no \ + markdown field — metadata-only fallback" + ); + } + markdown + } + Ok(resp) => { + tracing::warn!( + page_id = %page_id, + error = ?resp.error, + "[composio:notion] GET_PAGE_MARKDOWN failed — \ + metadata-only fallback" + ); + None + } + Err(e) => { + tracing::warn!( + page_id = %page_id, + error = %e, + "[composio:notion] GET_PAGE_MARKDOWN execute error — \ + metadata-only fallback" + ); + None + } + } + } + }) + .collect(); + + let bodies: Vec> = futures::stream::iter(body_futs) + .buffered(BODY_FETCH_CONCURRENCY) + .collect() + .await; + + // Count every request we fired (success or failure), matching the + // old per-call `record_requests(1)`. + state.record_requests(fetch_count as u32); + + for (p, body) in pending.iter_mut().zip(bodies.into_iter()) { + p.markdown_body = body; + } + } + + // ── Ingest queued pages (bounded concurrency) ─────────────────── + let ingestor = MemoryTreeIngestor { + config: ctx.config.as_ref(), + connection_id, + }; + let buffered = ingest_pending_buffered(&ingestor, pending, INGEST_CONCURRENCY).await; + IngestOutcome { + synced_keys: buffered.synced_keys, + persisted: buffered.persisted, + had_failures: buffered.had_failures, + } + } +} + +/// One page queued for concurrent ingest. Owns its raw page `Value` (the +/// orchestrator handed ownership via [`SyncItem`]). +struct PendingIngest { + sync_key: String, + page_id: String, + title: String, + edited_time: Option, + page: Value, + /// Rendered page body (markdown) fetched per-page via + /// `NOTION_GET_PAGE_MARKDOWN`. `None` when the body fetch was skipped + /// (budget) or failed — ingest falls back to the metadata-only body. + markdown_body: Option, +} + +/// Folded result of [`ingest_pending_buffered`]. Every field is +/// order-independent so the concurrent stage can accumulate regardless of the +/// order ingests complete. +#[derive(Default)] +struct BufferedIngestOutcome { + /// `sync_key`s whose ingest succeeded — the orchestrator marks each synced. + synced_keys: Vec, + /// Number of pages persisted (equals `synced_keys.len()`). + persisted: usize, + /// Whether any per-item ingest failed (the orchestrator holds the cursor). + had_failures: bool, +} + +/// Seam over "ingest one Notion page", so the bounded-concurrency driver can be +/// unit-tested with a fake that records peak in-flight calls without a real +/// memory tree or embedder. +#[async_trait] +trait PageIngestor { + async fn ingest( + &self, + page_id: &str, + title: &str, + edited_time: Option<&str>, + page: &Value, + markdown_body: Option<&str>, + ) -> anyhow::Result; +} + +/// Production ingestor: routes into the memory-tree pipeline (#2885) via +/// [`ingest_page_into_memory_tree`]. +struct MemoryTreeIngestor<'c> { + config: &'c Config, + connection_id: &'c str, +} + +#[async_trait] +impl PageIngestor for MemoryTreeIngestor<'_> { + async fn ingest( + &self, + page_id: &str, + title: &str, + edited_time: Option<&str>, + page: &Value, + markdown_body: Option<&str>, + ) -> anyhow::Result { + ingest_page_into_memory_tree( + self.config, + self.connection_id, + page_id, + title, + edited_time, + page, + markdown_body, + ) + .await + } +} + +/// Ingest the queued pages with bounded concurrency. Overlaps the per-item +/// embedding RTT (`buffer_unordered`, up to `concurrency` in flight) and folds +/// results into an order-independent [`BufferedIngestOutcome`]. Unordered is +/// correct here: nothing downstream depends on completion order — successes are +/// keyed by `sync_key`. +async fn ingest_pending_buffered( + ingestor: &I, + pending: Vec, + concurrency: usize, +) -> BufferedIngestOutcome { + let ingest_futs = pending + .into_iter() + .map(|p| async move { + let res = ingestor + .ingest( + &p.page_id, + &p.title, + p.edited_time.as_deref(), + &p.page, + p.markdown_body.as_deref(), + ) + .await; + (p.sync_key, p.page_id, res) + }) + .collect::>(); + + let mut outcome = BufferedIngestOutcome::default(); + let mut ingest_stream = futures::stream::iter(ingest_futs).buffer_unordered(concurrency); + while let Some((sync_key, page_id, res)) = ingest_stream.next().await { + match res { + Ok(_chunks_written) => { + outcome.synced_keys.push(sync_key); + outcome.persisted += 1; + } + Err(e) => { + outcome.had_failures = true; + tracing::warn!( + page_id = %page_id, + error = %e, + "[composio:notion] failed to ingest page into memory_tree (continuing)" + ); + } + } + } + outcome +} + +#[cfg(test)] +mod body_fetch_count_tests { + use super::body_fetch_count; + + #[test] + fn clamps_to_remaining_budget_when_budget_is_the_limit() { + // 10 pending pages but only 3 requests left today → fetch 3. + assert_eq!(body_fetch_count(10, 3), 3); + } + + #[test] + fn fetches_all_when_budget_exceeds_pending() { + assert_eq!(body_fetch_count(4, 100), 4); + } + + #[test] + fn zero_budget_fetches_nothing() { + // Mirrors the old serial "budget_exhausted() before first fetch" break. + assert_eq!(body_fetch_count(7, 0), 0); + } + + #[test] + fn zero_pending_fetches_nothing() { + assert_eq!(body_fetch_count(0, 50), 0); + } +} + +#[cfg(test)] +mod buffered_tests { + use super::*; + use serde_json::json; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + /// Fake ingestor: records the peak number of concurrent in-flight `ingest` + /// calls and can be told to fail one specific `page_id`. No memory tree or + /// embedder involved — lets us assert the concurrency bound and overlap + /// deterministically. + struct CountingIngestor { + in_flight: AtomicUsize, + peak: AtomicUsize, + fail_page: Option, + } + + impl CountingIngestor { + fn new(fail_page: Option<&str>) -> Arc { + Arc::new(Self { + in_flight: AtomicUsize::new(0), + peak: AtomicUsize::new(0), + fail_page: fail_page.map(str::to_string), + }) + } + } + + #[async_trait] + impl PageIngestor for CountingIngestor { + async fn ingest( + &self, + page_id: &str, + _title: &str, + _edited_time: Option<&str>, + _page: &Value, + _markdown_body: Option<&str>, + ) -> anyhow::Result { + let now = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1; + self.peak.fetch_max(now, Ordering::SeqCst); + for _ in 0..4 { + tokio::task::yield_now().await; + } + self.in_flight.fetch_sub(1, Ordering::SeqCst); + if self.fail_page.as_deref() == Some(page_id) { + Err(anyhow::anyhow!("forced failure for {page_id}")) + } else { + Ok(1) + } + } + } + + fn make_pending(n: usize) -> Vec { + (0..n) + .map(|i| PendingIngest { + sync_key: format!("k{i}"), + page_id: format!("p{i}"), + title: format!("Notion: page {i}"), + edited_time: None, + page: json!({ "id": format!("p{i}") }), + markdown_body: None, + }) + .collect() + } + + #[tokio::test] + async fn ingest_pending_buffered_bounds_and_overlaps() { + let ingestor = CountingIngestor::new(None); + let pending = make_pending(20); + + let outcome = ingest_pending_buffered(ingestor.as_ref(), pending, 8).await; + + assert_eq!(outcome.persisted, 20, "all pages persisted"); + assert_eq!(outcome.synced_keys.len(), 20); + assert!(!outcome.had_failures); + + let peak = ingestor.peak.load(Ordering::SeqCst); + assert!(peak <= 8, "peak in-flight {peak} exceeded the bound of 8"); + assert!(peak >= 2, "peak in-flight {peak} shows no real overlap"); + } + + #[tokio::test] + async fn ingest_pending_buffered_skips_failures_order_independent() { + let ingestor = CountingIngestor::new(Some("p2")); + let pending = make_pending(5); + + let outcome = ingest_pending_buffered(ingestor.as_ref(), pending, 4).await; + + assert_eq!(outcome.persisted, 4, "the one failed ingest is not counted"); + assert!(outcome.had_failures); + assert_eq!(outcome.synced_keys.len(), 4); + assert!( + !outcome.synced_keys.contains(&"k2".to_string()), + "the failed page's sync_key must not be marked synced" + ); + } + + #[test] + fn item_dedup_key_composes_id_and_edited_time() { + let with_edit = json!({ "id": "p1", "last_edited_time": "2026-05-01T00:00:00Z" }); + assert_eq!( + NotionSource.item_dedup_key(&with_edit).as_deref(), + Some("p1@2026-05-01T00:00:00Z") + ); + // No edited time → bare id. + let no_edit = json!({ "id": "p2" }); + assert_eq!(NotionSource.item_dedup_key(&no_edit).as_deref(), Some("p2")); + // No id at all → dropped. + assert_eq!( + NotionSource.item_dedup_key(&json!({ "last_edited_time": "x" })), + None + ); + } +} diff --git a/src/openhuman/memory_sync/composio/providers/orchestrator.rs b/src/openhuman/memory_sync/composio/providers/orchestrator.rs new file mode 100644 index 000000000..ad56a2da3 --- /dev/null +++ b/src/openhuman/memory_sync/composio/providers/orchestrator.rs @@ -0,0 +1,927 @@ +//! Generic incremental-sync orchestrator for Composio providers. +//! +//! Every Composio provider (gmail, notion, slack, …) used to hand-roll the +//! same ~15-step sync loop: load [`SyncState`] → check the daily budget → +//! resolve identity/scopes → page through results → dedupe against +//! `synced_ids` → drop items outside the `sync_depth_days` window → clamp to +//! `max_items` → ingest → advance the cursor → persist state → build a +//! [`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. +//! +//! ## Scopes: flat AND nested in one loop +//! +//! [`SyncScope`] is the abstraction that lets the same orchestrator drive both +//! shapes: +//! +//! * **Flat** providers (gmail, github, notion, linear) expose a *single +//! implicit scope* — [`SyncScope::flat`] — and page straight through their +//! one result stream. +//! * **Nested** providers (clickup workspaces → tasks, slack channels → +//! history) resolve their containers in [`IncrementalSource::preamble`] and +//! return one [`SyncScope`] per container; the orchestrator's +//! `for scope { for page {…} }` loop is byte-for-byte identical for both. +//! +//! Only Notion (flat) rides this path today, but the scope loop is written so +//! the nested providers slot in without a control-flow change — see the +//! migration checklist in the issue. + +use async_trait::async_trait; +use serde_json::{json, Value}; + +use super::helpers::ItemCap; +use super::sync_state::SyncState; +use super::{ProviderContext, SyncOutcome, SyncReason}; + +/// A unit of work to iterate within one sync pass. +/// +/// Flat providers return a single [`SyncScope::flat`]; nested providers return +/// one scope per container (workspace / channel). The orchestrator treats both +/// uniformly — that uniformity is the whole point of the abstraction. +#[derive(Debug, Clone)] +pub(crate) struct SyncScope { + /// Provider-native scope id (e.g. a ClickUp workspace id or a Slack + /// channel id). Empty string denotes the single implicit scope of a flat + /// provider. + pub id: String, + /// Human-readable label for logs. Never logged at a level that would leak + /// PII — ids/labels here are container identifiers, not user content. + pub label: String, +} + +impl SyncScope { + /// The single implicit scope of a flat provider (gmail/github/notion/linear). + pub(crate) fn flat() -> Self { + Self { + id: String::new(), + label: "".to_string(), + } + } + + /// One container of a nested provider (clickup workspace, slack channel). + pub(crate) fn nested(id: impl Into, label: impl Into) -> Self { + Self { + id: id.into(), + label: label.into(), + } + } +} + +/// One page fetched from a provider for a given [`SyncScope`]. +pub(crate) struct PageFetch { + /// Raw upstream items, already unwrapped from the Composio envelope by the + /// provider's [`IncrementalSource::fetch_page`]. + pub items: Vec, + /// Opaque next-page cursor for *this scope*, or `None` when this was the + /// last page. The orchestrator never interprets it — it is a Notion + /// `start_cursor`, a ClickUp page index, etc., round-tripped back into the + /// next `fetch_page` call. + pub next: Option, +} + +/// One item that survived dedup + the depth window + the `max_items` clamp and +/// is queued for ingest. +pub(crate) struct SyncItem { + /// Stable dedup key (e.g. `{page_id}@{edited_time}`). Marked synced on a + /// successful ingest so the next pass skips it. + pub dedup_key: String, + /// Sort timestamp used for cursor advancement and depth-window compares. + /// Same representation as [`IncrementalSource::depth_floor`]. + pub sort_ts: Option, + /// Raw upstream payload, handed to [`IncrementalSource::ingest`]. + pub raw: Value, +} + +/// Folded result of one [`IncrementalSource::ingest`] call. Every field is +/// order-independent so a concurrent ingest stage can accumulate into it. +#[derive(Default)] +pub(crate) struct IngestOutcome { + /// Dedup keys whose ingest succeeded — the orchestrator marks each synced. + pub synced_keys: Vec, + /// Number of items persisted (equals `synced_keys.len()`). + pub persisted: usize, + /// Whether any per-item ingest failed. When true and the source opts into + /// [`IncrementalSource::hold_cursor_on_ingest_failure`], the orchestrator + /// holds the cursor so the failed range is re-fetched next pass. + pub had_failures: bool, +} + +/// The slim primitive a Composio provider implements to ride [`run_sync`]. +/// +/// Implementations own *only* the provider-specific shapes — which actions to +/// call, how to read ids/timestamps, how to persist. The orchestrator owns all +/// the control flow (budget, pagination bound, dedup, depth window, cap, cursor +/// advance, state persistence). +#[async_trait] +pub(crate) trait IncrementalSource: Send + Sync { + /// Toolkit slug — used for [`SyncState`] keying and log prefixes. + fn toolkit(&self) -> &'static str; + + /// Page size to request this pass. Providers typically widen this for + /// [`SyncReason::ConnectionCreated`] to backfill faster. + fn page_size(&self, reason: SyncReason) -> u32; + + /// The provider's own internal page ceiling — the `fallback` handed to + /// [`ItemCap::max_pages`], applied *per scope*. + fn max_pages(&self) -> u32; + + /// Resolve identity and list the scopes to iterate. + /// + /// Flat providers return `vec![SyncScope::flat()]`. Nested providers call + /// their "list workspaces / channels" action(s) here and return one scope + /// each (recording any budget spent via `state`). Returning an empty vec + /// short-circuits the pass to a no-op outcome. + async fn preamble( + &self, + ctx: &ProviderContext, + state: &mut SyncState, + ) -> Result, String>; + + /// Fetch one page of raw items for `scope` at `cursor` (`None` = first + /// page). Return the items already unwrapped from the Composio envelope + /// plus the opaque next-page token. + /// + /// Implementations **must record the page request against the daily budget** + /// (`state.record_requests(1)`) for any *completed* round-trip — including + /// one the provider reports as `successful == false` — before converting it + /// to an `Err`, so a broken/unauthorized connection cannot make unlimited + /// billable failed calls without hitting the per-day cap. A transport error + /// (no round-trip) must not be recorded. + async fn fetch_page( + &self, + ctx: &ProviderContext, + scope: &SyncScope, + cursor: Option<&str>, + reason: SyncReason, + state: &mut SyncState, + ) -> Result; + + /// Stable dedup key for one raw item. `None` drops the item (e.g. it has no + /// extractable id). + fn item_dedup_key(&self, item: &Value) -> Option; + + /// Sort timestamp for one raw item — compared against the persistent cursor + /// and the depth floor. `None` means "no timestamp" (never trips the cursor + /// boundary or the depth window). + fn item_sort_ts(&self, item: &Value) -> Option; + + /// Build the `sync_depth_days` floor in the *same representation* as + /// [`Self::item_sort_ts`] so the lexicographic compare is valid. Default is + /// RFC3339 UTC; providers whose timestamps are epoch-millis strings + /// (clickup) override. + fn depth_floor(&self, days: u32) -> String { + let floor = chrono::Utc::now() - chrono::Duration::days(days as i64); + floor.to_rfc3339() + } + + /// Whether to hold (not advance) the cursor when an ingest reported a + /// failure this pass. Default `true` — Notion's safe behaviour under the + /// delete-first memory-tree pipeline (#2885), where an edited item that + /// fails to re-ingest must be re-fetched. Providers that advance regardless + /// of per-item failures (clickup) override to `false`. + fn hold_cursor_on_ingest_failure(&self) -> bool { + true + } + + /// Persist a batch of already-filtered items. May spend budget via `state` + /// (e.g. Notion's per-page body fetch). Returns which dedup keys succeeded + /// so the orchestrator can mark them synced. + async fn ingest( + &self, + ctx: &ProviderContext, + scope: &SyncScope, + state: &mut SyncState, + items: Vec, + ) -> IngestOutcome; +} + +/// Current wall-clock time in milliseconds since the UNIX epoch. +fn now_ms() -> u64 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + +/// Build the "nothing to do" outcome used by the budget-exhausted and +/// empty-scopes early returns. +fn skipped_outcome( + toolkit: &str, + connection_id: &str, + reason: SyncReason, + started_at_ms: u64, + why: &str, + details: Value, +) -> SyncOutcome { + SyncOutcome { + toolkit: toolkit.to_string(), + connection_id: Some(connection_id.to_string()), + reason: reason.as_str().to_string(), + items_ingested: 0, + started_at_ms, + finished_at_ms: now_ms(), + summary: format!("{toolkit} sync skipped: {why}"), + details, + } +} + +/// Pure (no I/O) per-page scan: extract dedup key + sort timestamp for each raw +/// item, advance `newest_ts`, detect the persistent-cursor boundary, and drop +/// already-synced items. Returns the survivors plus whether we crossed the +/// cursor boundary (the signal to stop paginating this scope). +/// +/// This is the generic form of every provider's old `select_pending`. All +/// order-dependent decisions live here so the (possibly concurrent) ingest +/// stage never has to reason about ordering. +fn select_pending( + source: &S, + items: &[Value], + state: &SyncState, + newest_ts: &mut Option, +) -> (Vec, bool) { + let mut hit_cursor_boundary = false; + let mut pending: Vec = Vec::new(); + for item in items { + let Some(dedup_key) = source.item_dedup_key(item) else { + tracing::debug!( + toolkit = source.toolkit(), + "[composio:sync_orch] item missing dedup key, skipping" + ); + continue; + }; + + let sort_ts = source.item_sort_ts(item); + + // Track the newest timestamp for cursor advancement — for *every* item + // with a timestamp, including ones we skip as already-synced. + if let Some(ref ts) = sort_ts { + if newest_ts.as_ref().is_none_or(|existing| ts > existing) { + *newest_ts = Some(ts.clone()); + } + } + + // Older-or-equal to the cursor AND already synced → we have caught up. + if let (Some(cursor), Some(ts)) = (&state.cursor, &sort_ts) { + if ts <= cursor && state.is_synced(&dedup_key) { + hit_cursor_boundary = true; + continue; + } + } + + if state.is_synced(&dedup_key) { + continue; + } + + pending.push(SyncItem { + dedup_key, + sort_ts, + raw: item.clone(), + }); + } + (pending, hit_cursor_boundary) +} + +/// Run one incremental sync end-to-end through the generic orchestrator. +/// +/// The provider supplies the [`IncrementalSource`] primitives; everything else +/// — budget, the per-scope page loop bounded by [`ItemCap::max_pages`], dedup, +/// the `sync_depth_days` window, the precise `max_items` clamp, cursor +/// advance/hold, state persistence, and the [`SyncOutcome`] — is owned here. +pub(crate) async fn run_sync( + source: &S, + ctx: &ProviderContext, + reason: SyncReason, +) -> Result { + let toolkit = source.toolkit(); + let started_at_ms = now_ms(); + let connection_id = ctx + .connection_id + .clone() + .unwrap_or_else(|| "default".to_string()); + + tracing::info!( + toolkit, + connection_id = %connection_id, + reason = reason.as_str(), + "[composio:sync_orch] incremental sync starting" + ); + + // ── Step 1: load persistent sync state ────────────────────────────── + let Some(memory) = ctx.memory_client() else { + return Err(format!("[composio:{toolkit}] memory client not ready")); + }; + let mut state = SyncState::load(&memory, toolkit, &connection_id).await?; + + // ── Step 2: daily budget pre-check ────────────────────────────────── + if state.budget_exhausted() { + tracing::info!( + toolkit, + connection_id = %connection_id, + "[composio:sync_orch] daily request budget exhausted, skipping sync" + ); + return Ok(skipped_outcome( + toolkit, + &connection_id, + reason, + started_at_ms, + "daily budget exhausted", + json!({ "budget_exhausted": true }), + )); + } + + // ── Step 3: preamble — resolve identity + scopes ──────────────────── + let scopes = match source.preamble(ctx, &mut state).await { + Ok(scopes) => scopes, + Err(e) => { + // Persist any budget spent during the preamble before propagating. + let _ = state.save(&memory).await; + return Err(e); + } + }; + + if scopes.is_empty() { + tracing::info!( + toolkit, + connection_id = %connection_id, + "[composio:sync_orch] no scopes to sync" + ); + state.save(&memory).await?; + return Ok(skipped_outcome( + toolkit, + &connection_id, + reason, + started_at_ms, + "no scopes to sync", + json!({ "scopes": 0 }), + )); + } + + // ── Step 4: caps + window ─────────────────────────────────────────── + let page_size = source.page_size(reason); + let mut cap = ItemCap::new(ctx.max_items); + let effective_max_pages = cap.max_pages(page_size, source.max_pages()); + if ctx.max_items.is_some() && effective_max_pages < source.max_pages() { + tracing::debug!( + toolkit, + connection_id = %connection_id, + max_items = ?ctx.max_items, + effective_max_pages, + "[composio:sync_orch] [memory_sync] applying max_items page cap" + ); + } + + let depth_floor: Option = ctx.sync_depth_days.map(|days| { + let floor = source.depth_floor(days); + tracing::debug!( + toolkit, + connection_id = %connection_id, + sync_depth_days = days, + oldest_allowed = %floor, + "[composio:sync_orch] [memory_sync] applying sync_depth_days floor" + ); + floor + }); + + // ── Step 5: scope × page loop ─────────────────────────────────────── + let mut total_fetched: usize = 0; + let mut total_persisted: usize = 0; + let mut newest_ts: Option = None; + let mut had_ingest_failures = false; + let mut hit_cap_boundary = false; + + 'scopes: for scope in &scopes { + // The page cursor is per-scope — reset at the top of every scope. + let mut cursor: Option = None; + + for page_num in 0..effective_max_pages { + if state.budget_exhausted() { + tracing::info!( + toolkit, + scope = %scope.label, + page = page_num, + "[composio:sync_orch] budget exhausted mid-sync, stopping" + ); + break 'scopes; + } + + // `fetch_page` records the page request against the budget (incl. + // provider-reported failures) per its contract. On error we persist + // whatever budget/dedup progress we have before propagating — + // parity with the per-provider loops, which saved state before + // returning a failed-page error. + let fetch = match source + .fetch_page(ctx, scope, cursor.as_deref(), reason, &mut state) + .await + { + Ok(fetch) => fetch, + Err(e) => { + let _ = state.save(&memory).await; + return Err(e); + } + }; + total_fetched += fetch.items.len(); + + if fetch.items.is_empty() { + tracing::debug!( + toolkit, + scope = %scope.label, + page = page_num, + "[composio:sync_orch] empty page, moving on" + ); + break; + } + + // Dedup + cursor-boundary detection + newest-ts tracking. + let (mut pending, mut hit_cursor_boundary) = + select_pending(source, &fetch.items, &state, &mut newest_ts); + + // sync_depth_days: `pending` is in descending-timestamp order, so + // truncate at the first item below the floor and stop paginating. + if let Some(ref floor) = depth_floor { + if let Some(cut) = pending.iter().position(|it| { + it.sort_ts + .as_deref() + .map(|t| t < floor.as_str()) + .unwrap_or(false) + }) { + pending.truncate(cut); + hit_cursor_boundary = true; + } + } + + // max_items: clamp the dedup'd batch to the remaining budget BEFORE + // ingest — the precise cap that fixes the page-granular #3304 bug. + cap.clamp_batch(&mut pending); + + // Provider-specific persistence (may spend budget, e.g. body fetch). + let outcome = source.ingest(ctx, scope, &mut state, pending).await; + for key in &outcome.synced_keys { + state.mark_synced(key); + } + total_persisted += outcome.persisted; + cap.record(outcome.persisted); + if outcome.had_failures { + had_ingest_failures = true; + } + + // Precise cap reached → stop the entire pass. + if cap.is_reached() { + hit_cap_boundary = true; + break 'scopes; + } + + if hit_cursor_boundary { + tracing::debug!( + toolkit, + scope = %scope.label, + page = page_num, + "[composio:sync_orch] reached cursor/depth boundary, stopping scope" + ); + break; + } + + cursor = fetch.next; + if cursor.is_none() { + tracing::debug!( + toolkit, + scope = %scope.label, + page = page_num, + "[composio:sync_orch] no next cursor, scope done" + ); + break; + } + } + } + + // ── Step 6: advance cursor (or hold) and persist state ────────────── + // + // Hold the cursor on a cap-truncated pass so the next sync re-scans the + // unseen tail, and on an ingest failure when the source opts in (Notion's + // delete-first safety). Otherwise advance to the newest timestamp seen. + let hold_cursor = + hit_cap_boundary || (had_ingest_failures && source.hold_cursor_on_ingest_failure()); + if !hold_cursor { + if let Some(new_cursor) = newest_ts { + state.advance_cursor(&new_cursor); + } + } else { + tracing::warn!( + toolkit, + connection_id = %connection_id, + had_ingest_failures, + hit_cap_boundary, + "[composio:sync_orch] holding cursor — cap-truncated pass or ingest failures; \ + next sync will re-fetch the unseen/failed range" + ); + } + state.set_last_sync_at_ms(now_ms()); + state.save(&memory).await?; + + let finished_at_ms = now_ms(); + let summary = format!( + "{toolkit} sync ({reason}): fetched {total_fetched}, persisted {total_persisted} new, \ + budget remaining {remaining}", + reason = reason.as_str(), + remaining = state.budget_remaining(), + ); + tracing::info!( + toolkit, + connection_id = %connection_id, + elapsed_ms = finished_at_ms.saturating_sub(started_at_ms), + total_fetched, + total_persisted, + budget_remaining = state.budget_remaining(), + "[composio:sync_orch] incremental sync complete" + ); + + Ok(SyncOutcome { + toolkit: toolkit.to_string(), + connection_id: Some(connection_id), + reason: reason.as_str().to_string(), + items_ingested: total_persisted, + started_at_ms, + finished_at_ms, + summary, + details: json!({ + "results_fetched": total_fetched, + "results_persisted": total_persisted, + "budget_remaining": state.budget_remaining(), + "cursor": state.cursor, + "synced_ids_total": state.synced_ids.len(), + }), + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::openhuman::config::Config; + use std::sync::Arc; + use tempfile::TempDir; + + /// A minimal in-test [`IncrementalSource`] that proves a *future* toolkit + /// inherits the cap + window for free. It synthesises items per scope (or + /// returns explicit ones) and "ingests" by counting — no real memory tree. + /// `Default` keeps the per-test literals to just the field(s) they vary. + #[derive(Default)] + struct FakeSource { + scopes: Vec, + items_per_scope: usize, + /// When set, returned verbatim for the first page of the (single) scope + /// instead of synthesised items — used by the depth-window test. + explicit_items: Option>, + /// When true, `preamble` returns an error (exercises the preamble-error + /// save-and-propagate path). + fail_preamble: bool, + /// When true, `fetch_page` returns a *transport* error (no round-trip → + /// not budget-recorded). + fail_fetch: bool, + /// When true, `fetch_page` records the round-trip *then* returns a + /// provider-reported failure — pins that a failed page still consumes + /// the daily budget. + provider_fail_fetch: bool, + } + + impl FakeSource { + fn flat(items_per_scope: usize) -> Self { + Self { + scopes: vec![SyncScope::flat()], + items_per_scope, + ..Default::default() + } + } + } + + #[async_trait] + impl IncrementalSource for FakeSource { + fn toolkit(&self) -> &'static str { + "faketoolkit" + } + fn page_size(&self, _reason: SyncReason) -> u32 { + 50 + } + fn max_pages(&self) -> u32 { + 20 + } + async fn preamble( + &self, + _ctx: &ProviderContext, + state: &mut SyncState, + ) -> Result, String> { + if self.fail_preamble { + // Spend a request first so the save-on-error path persists it. + state.record_requests(1); + return Err("fake preamble failure".to_string()); + } + Ok(self.scopes.clone()) + } + async fn fetch_page( + &self, + _ctx: &ProviderContext, + scope: &SyncScope, + cursor: Option<&str>, + _reason: SyncReason, + state: &mut SyncState, + ) -> Result { + if self.fail_fetch { + // Simulate a transport error (no completed round-trip) → not + // recorded, matching the contract. + return Err("fake fetch_page failure".to_string()); + } + // A completed round-trip — record it against the budget. + state.record_requests(1); + if self.provider_fail_fetch { + // Completed but the provider reported failure — already recorded. + return Err("fake provider-reported page failure".to_string()); + } + // Single page per scope: everything comes back on the first call. + if cursor.is_some() { + return Ok(PageFetch { + items: vec![], + next: None, + }); + } + if let Some(items) = &self.explicit_items { + return Ok(PageFetch { + items: items.clone(), + next: None, + }); + } + let items = (0..self.items_per_scope) + .map(|i| { + json!({ + "id": format!("{}-{i}", scope.id), + "ts": "2099-01-01T00:00:00Z" + }) + }) + .collect(); + Ok(PageFetch { items, next: None }) + } + fn item_dedup_key(&self, item: &Value) -> Option { + item.get("id").and_then(Value::as_str).map(str::to_string) + } + fn item_sort_ts(&self, item: &Value) -> Option { + item.get("ts").and_then(Value::as_str).map(str::to_string) + } + async fn ingest( + &self, + _ctx: &ProviderContext, + _scope: &SyncScope, + _state: &mut SyncState, + items: Vec, + ) -> IngestOutcome { + let synced_keys: Vec = items.into_iter().map(|it| it.dedup_key).collect(); + let persisted = synced_keys.len(); + IngestOutcome { + synced_keys, + persisted, + had_failures: false, + } + } + } + + fn fake_ctx( + tmp: &TempDir, + max_items: Option, + sync_depth_days: Option, + ) -> ProviderContext { + let mut config = Config { + config_path: tmp.path().join("config.toml"), + workspace_dir: tmp.path().join("workspace"), + ..Config::default() + }; + config.secrets.encrypt = false; + ProviderContext { + config: Arc::new(config), + toolkit: "faketoolkit".to_string(), + connection_id: Some("conn-fake".to_string()), + usage: Default::default(), + max_items, + sync_depth_days, + } + } + + #[tokio::test] + async fn max_items_caps_ingest_to_exact_count_not_page_granular() { + let tmp = TempDir::new().unwrap(); + let ctx = fake_ctx(&tmp, Some(2), None); + // One page returns 5 items; the cap is 2. + let outcome = run_sync(&FakeSource::flat(5), &ctx, SyncReason::ConnectionCreated) + .await + .expect("run_sync"); + assert_eq!( + outcome.items_ingested, 2, + "max_items=2 must clamp a 5-item page to EXACTLY 2 (the #3304 fix)" + ); + } + + #[tokio::test] + async fn no_cap_ingests_the_full_page() { + let tmp = TempDir::new().unwrap(); + let ctx = fake_ctx(&tmp, None, None); + let outcome = run_sync(&FakeSource::flat(5), &ctx, SyncReason::Periodic) + .await + .expect("run_sync"); + assert_eq!( + outcome.items_ingested, 5, + "with no cap every valid page item is ingested" + ); + } + + #[tokio::test] + async fn sync_depth_days_filters_items_below_the_floor() { + let tmp = TempDir::new().unwrap(); + let ctx = fake_ctx(&tmp, None, Some(7)); + // Descending timestamp order: two recent (far future), three ancient. + // With a 7-day floor only the two recent items survive. + let items = vec![ + json!({ "id": "a", "ts": "2099-01-02T00:00:00Z" }), + json!({ "id": "b", "ts": "2099-01-01T00:00:00Z" }), + json!({ "id": "c", "ts": "2000-01-03T00:00:00Z" }), + json!({ "id": "d", "ts": "2000-01-02T00:00:00Z" }), + json!({ "id": "e", "ts": "2000-01-01T00:00:00Z" }), + ]; + let source = FakeSource { + scopes: vec![SyncScope::flat()], + explicit_items: Some(items), + ..Default::default() + }; + let outcome = run_sync(&source, &ctx, SyncReason::Manual) + .await + .expect("run_sync"); + assert_eq!( + outcome.items_ingested, 2, + "sync_depth_days=7 must drop the three ancient items" + ); + } + + #[tokio::test] + async fn nested_scopes_share_one_cap_budget() { + let tmp = TempDir::new().unwrap(); + let ctx = fake_ctx(&tmp, Some(4), None); + // Two scopes, 3 items each (6 total); the cap is 4 → 3 from scope one, + // 1 from scope two, then the pass stops. Proves the cap spans scopes. + let source = FakeSource { + scopes: vec![ + SyncScope::nested("s1", "Scope 1"), + SyncScope::nested("s2", "Scope 2"), + ], + items_per_scope: 3, + ..Default::default() + }; + let outcome = run_sync(&source, &ctx, SyncReason::ConnectionCreated) + .await + .expect("run_sync"); + assert_eq!( + outcome.items_ingested, 4, + "max_items must cap the combined ingest across nested scopes" + ); + } + + #[tokio::test] + async fn budget_exhausted_short_circuits_to_a_skip_outcome() { + let tmp = TempDir::new().unwrap(); + let ctx = fake_ctx(&tmp, None, None); + // Drain the daily budget before the run so the pre-check trips. + { + let memory = ctx.memory_client().expect("memory client"); + let mut state = SyncState::load(&memory, "faketoolkit", "conn-fake") + .await + .unwrap(); + state.record_requests(state.budget_remaining()); + state.save(&memory).await.unwrap(); + } + let outcome = run_sync(&FakeSource::flat(5), &ctx, SyncReason::Periodic) + .await + .expect("run_sync"); + assert_eq!(outcome.items_ingested, 0); + assert!( + outcome.summary.contains("budget"), + "exhausted-budget run must report a skip, got: {}", + outcome.summary + ); + } + + #[tokio::test] + async fn empty_scopes_short_circuit_to_a_skip_outcome() { + let tmp = TempDir::new().unwrap(); + let ctx = fake_ctx(&tmp, None, None); + let source = FakeSource { + scopes: vec![], // preamble resolved no scopes to iterate + items_per_scope: 5, + ..Default::default() + }; + let outcome = run_sync(&source, &ctx, SyncReason::Periodic) + .await + .expect("run_sync"); + assert_eq!(outcome.items_ingested, 0); + assert!( + outcome.summary.contains("no scopes"), + "empty scopes must report a skip, got: {}", + outcome.summary + ); + } + + #[tokio::test] + async fn preamble_error_propagates() { + let tmp = TempDir::new().unwrap(); + let ctx = fake_ctx(&tmp, None, None); + let source = FakeSource { + scopes: vec![SyncScope::flat()], + items_per_scope: 5, + fail_preamble: true, + ..Default::default() + }; + let err = run_sync(&source, &ctx, SyncReason::Periodic) + .await + .expect_err("preamble failure must propagate"); + assert!(err.contains("preamble"), "got: {err}"); + } + + #[tokio::test] + async fn fetch_page_error_propagates() { + let tmp = TempDir::new().unwrap(); + let ctx = fake_ctx(&tmp, None, None); + let source = FakeSource { + scopes: vec![SyncScope::flat()], + items_per_scope: 5, + fail_fetch: true, + ..Default::default() + }; + let err = run_sync(&source, &ctx, SyncReason::Periodic) + .await + .expect_err("fetch_page failure must propagate"); + assert!(err.contains("fetch_page"), "got: {err}"); + } + + #[tokio::test] + async fn provider_reported_page_failure_still_consumes_budget() { + // Parity with the per-provider loops (and the Codex review): a page that + // completes the round-trip but reports `successful == false` must count + // against the daily budget before the error propagates, so a broken + // connection can't make unlimited billable failed page calls. + let tmp = TempDir::new().unwrap(); + let ctx = fake_ctx(&tmp, None, None); + let source = FakeSource { + scopes: vec![SyncScope::flat()], + items_per_scope: 5, + provider_fail_fetch: true, + ..Default::default() + }; + let before = { + let memory = ctx.memory_client().expect("memory client"); + SyncState::load(&memory, "faketoolkit", "conn-fake") + .await + .unwrap() + .budget_remaining() + }; + let err = run_sync(&source, &ctx, SyncReason::Periodic) + .await + .expect_err("provider-reported failure must propagate"); + assert!(err.contains("provider-reported"), "got: {err}"); + // The orchestrator saved state on the page error; the failed page was + // recorded, so exactly one request was consumed. + let memory = ctx.memory_client().expect("memory client"); + let after = SyncState::load(&memory, "faketoolkit", "conn-fake") + .await + .unwrap() + .budget_remaining(); + assert_eq!( + before - after, + 1, + "a completed-but-failed page must consume exactly one budget request" + ); + } + + #[test] + fn select_pending_tracks_newest_skips_synced_and_detects_boundary() { + let source = FakeSource::flat(0); + let mut state = SyncState::new("faketoolkit", "conn1"); + state.cursor = Some("2026-04-15T00:00:00Z".to_string()); + // Item B is already synced and older than the cursor. + state.mark_synced("b"); + + let items = vec![ + json!({ "id": "a", "ts": "2026-05-01T00:00:00Z" }), + json!({ "id": "b", "ts": "2026-04-01T00:00:00Z" }), + json!({ "ts": "2026-03-01T00:00:00Z" }), // no id → dropped + ]; + + let mut newest: Option = None; + let (pending, hit_boundary) = select_pending(&source, &items, &state, &mut newest); + + assert_eq!(pending.len(), 1, "only the new item A survives"); + assert_eq!(pending[0].dedup_key, "a"); + assert!( + hit_boundary, + "older synced item B trips the cursor boundary" + ); + assert_eq!(newest.as_deref(), Some("2026-05-01T00:00:00Z")); + } +} diff --git a/tests/memory_sync_providers_raw_coverage_e2e.rs b/tests/memory_sync_providers_raw_coverage_e2e.rs index 9cc9c9fd6..2f8a4bb54 100644 --- a/tests/memory_sync_providers_raw_coverage_e2e.rs +++ b/tests/memory_sync_providers_raw_coverage_e2e.rs @@ -1102,6 +1102,87 @@ async fn notion_sync_max_items_caps_ingest_to_exact_count() { server.abort(); } +// ───────────────────────────────────────────────────────────────────────── +// Notion sync_depth_days enforcement (via the shared orchestrator) +// +// Proves the generic orchestrator's depth window actually drops items older +// than the floor end-to-end through the real Notion provider. The mock returns +// 2 recent pages (far-future `last_edited_time`, always inside the window) and +// 3 ancient pages (year-2000, always outside it), in the descending order the +// provider requests. With sync_depth_days=7 only the 2 recent pages persist — +// the orchestrator truncates the page at the first item below the floor. +// ───────────────────────────────────────────────────────────────────────── + +/// Build `recent` + `old` Notion pages in descending `last_edited_time` order. +/// Recent pages use a far-future timestamp (always within any depth window); +/// old pages use a year-2000 timestamp (always outside it). +fn notion_depth_pages(recent: usize, old: usize) -> Vec { + let mut pages = Vec::new(); + for i in 0..recent { + pages.push(json!({ + "id": format!("notion-recent-{i:04}"), + "object": "page", + "last_edited_time": format!("2099-12-{:02}T10:00:00.000Z", 28 - i), + "properties": { "Name": { "type": "title", "title": [{ "plain_text": format!("Recent {i}") }] } } + })); + } + for i in 0..old { + pages.push(json!({ + "id": format!("notion-old-{i:04}"), + "object": "page", + "last_edited_time": format!("2000-01-{:02}T10:00:00.000Z", 3 - i), + "properties": { "Name": { "type": "title", "title": [{ "plain_text": format!("Old {i}") }] } } + })); + } + pages +} + +#[tokio::test] +async fn notion_sync_depth_days_filters_old_pages() { + let _guard = env_lock(); + let tmp = TempDir::new().expect("tempdir"); + let _workspace = EnvGuard::set_path("OPENHUMAN_WORKSPACE", tmp.path()); + let _home = EnvGuard::set_path("HOME", tmp.path()); + let _backend = EnvGuard::unset("BACKEND_URL"); + + // One page: 2 recent + 3 ancient. With a 7-day window only the 2 recent + // pages must be ingested. + let requests: Arc>> = Arc::new(Mutex::new(Vec::new())); + let (base, server) = loopback_router(notion_cap_router( + notion_depth_pages(2, 3), + Arc::clone(&requests), + )) + .await; + + let mut config = config_in(&tmp); + config.api_url = Some(base); + persist_config(&config).await; + store_session(&config); + memory_global::init(config.workspace_dir.clone()).expect("init global memory"); + + let ctx = ProviderContext { + config: Arc::new(config), + toolkit: "notion".to_string(), + connection_id: Some("conn-notion-depth".to_string()), + usage: Default::default(), + max_items: None, + // 7-day window: drops the year-2000 pages, keeps the far-future ones. + sync_depth_days: Some(7), + }; + + let outcome = NotionProvider::new() + .sync(&ctx, SyncReason::ConnectionCreated) + .await + .expect("notion depth sync"); + + assert_eq!( + outcome.items_ingested, 2, + "sync_depth_days=7 must drop the 3 year-2000 pages and keep the 2 recent ones" + ); + + server.abort(); +} + // ───────────────────────────────────────────────────────────────────────── // Linear cap enforcement //