admin/openhuman
1
0
Fork 0
mirror of https://github.com/tinyhumansai/openhuman.git synced 2026-07-27 21:08:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(memory_sync): generic Composio sync orchestrator + convert Notion (Refs #3333) (#3852)

Browse Source 
Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
sanil-23
2026-06-21 13:25:22 -07:00
committed by GitHub
co-authored by Claude
parent b9d2cdd735
commit 7758148c2c
6 changed files with 1570 additions and 754 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openhuman/memory_sync/composio/providers/mod.rs
+1
View File
@@ -34,6 +34,7 @@
mod descriptions;
pub(crate) mod helpers;
pub(crate) mod orchestrator;
mod scope_lookup;
pub mod tool_scope;
mod traits;
src/openhuman/memory_sync/composio/providers/notion/mod.rs
+1
View File
@@ -1,5 +1,6 @@
mod ingest;
mod provider;
mod source;
mod sync;
#[cfg(test)]
mod tests;
src/openhuman/memory_sync/composio/providers/notion/provider.rs
+10 -754
View File
@@ -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<SyncOutcome, String> {
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<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;
let mut notion_cursor: Option<String> = 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<String> = 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::<Vec<_>>()),
"[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<Option<String>> = 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<String> {
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<String>,
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<String>,
}
/// 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<String>,
/// 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<usize>;
}
/// 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<usize> {
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<String>,
) -> (Vec<PendingIngest<'a>>, bool) {
let mut hit_cursor_boundary = false;
let mut pending: Vec<PendingIngest> = 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<I: PageIngestor + Sync>(
ingestor: &I,
pending: Vec<PendingIngest<'_>>,
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::<Vec<_>>();
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<String>,
}
impl CountingIngestor {
fn new(fail_page: Option<&str>) -> Arc<Self> {
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<usize> {
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<Value> {
(0..n).map(|i| json!({ "id": format!("p{i}") })).collect()
}
fn make_pending<'a>(pages: &'a [Value]) -> Vec<PendingIngest<'a>> {
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<String> = 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"));
}
}
src/openhuman/memory_sync/composio/providers/notion/source.rs
+550
View File
@@ -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<SyncOutcome, String> {
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<Vec<SyncScope>, String> {
Ok(vec![SyncScope::flat()])
}
async fn fetch_page(
&self,
ctx: &ProviderContext,
_scope: &SyncScope,
cursor: Option<&str>,
reason: SyncReason,
state: &mut SyncState,
) -> Result<PageFetch, String> {
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<String> {
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<String> {
extract_item_id(item, PAGE_EDITED_PATHS)
}
async fn ingest(
&self,
ctx: &ProviderContext,
_scope: &SyncScope,
state: &mut SyncState,
items: Vec<SyncItem>,
) -> 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<PendingIngest> = 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<String> = 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<Option<String>> = 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<String>,
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<String>,
}
/// 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<String>,
/// 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<usize>;
}
/// 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<usize> {
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<I: PageIngestor + Sync>(
ingestor: &I,
pending: Vec<PendingIngest>,
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::<Vec<_>>();
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<String>,
}
impl CountingIngestor {
fn new(fail_page: Option<&str>) -> Arc<Self> {
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<usize> {
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<PendingIngest> {
(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
);
}
}
src/openhuman/memory_sync/composio/providers/orchestrator.rs
+927
View File
@@ -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: "<flat>".to_string(),
}
}
/// One container of a nested provider (clickup workspace, slack channel).
pub(crate) fn nested(id: impl Into<String>, label: impl Into<String>) -> 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<Value>,
/// 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<String>,
}
/// 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<String>,
/// 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<String>,
/// 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<Vec<SyncScope>, 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<PageFetch, String>;
/// 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<String>;
/// 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<String>;
/// 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<SyncItem>,
) -> 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<S: IncrementalSource + ?Sized>(
source: &S,
items: &[Value],
state: &SyncState,
newest_ts: &mut Option<String>,
) -> (Vec<SyncItem>, bool) {
let mut hit_cursor_boundary = false;
let mut pending: Vec<SyncItem> = 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<S: IncrementalSource + ?Sized>(
source: &S,
ctx: &ProviderContext,
reason: SyncReason,
) -> Result<SyncOutcome, String> {
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<String> = 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<String> = 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<String> = 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<SyncScope>,
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<Vec<Value>>,
/// 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<Vec<SyncScope>, 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<PageFetch, String> {
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<String> {
item.get("id").and_then(Value::as_str).map(str::to_string)
}
fn item_sort_ts(&self, item: &Value) -> Option<String> {
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<SyncItem>,
) -> IngestOutcome {
let synced_keys: Vec<String> = 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<u32>,
sync_depth_days: Option<u32>,
) -> 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<String> = 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"));
}
}
tests/memory_sync_providers_raw_coverage_e2e.rs
+81
View File
@@ -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<Value> {
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<Mutex<Vec<Value>>> = 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
//