diff --git a/src/openhuman/memory_sync/composio/providers/github/mod.rs b/src/openhuman/memory_sync/composio/providers/github/mod.rs index a6aac2a89..a95357bb6 100644 --- a/src/openhuman/memory_sync/composio/providers/github/mod.rs +++ b/src/openhuman/memory_sync/composio/providers/github/mod.rs @@ -14,6 +14,7 @@ mod ingest; mod provider; +mod source; mod sync; #[cfg(test)] mod tests; diff --git a/src/openhuman/memory_sync/composio/providers/github/provider.rs b/src/openhuman/memory_sync/composio/providers/github/provider.rs index 42749b8c8..9f55ec028 100644 --- a/src/openhuman/memory_sync/composio/providers/github/provider.rs +++ b/src/openhuman/memory_sync/composio/providers/github/provider.rs @@ -23,9 +23,8 @@ use async_trait::async_trait; use serde_json::{json, Value}; use std::time::Duration; -use super::ingest::ingest_issue_into_memory_tree; +use super::source::run_github_sync; use super::sync; -use crate::openhuman::memory_sync::composio::providers::sync_state::SyncState; use crate::openhuman::memory_sync::composio::providers::{ merge_extra, pick_str, resolve_sync_interval_secs, ComposioProvider, CuratedTool, GithubFetchMode, NormalizedTask, ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason, @@ -35,16 +34,6 @@ use crate::openhuman::memory_sync::composio::providers::{ pub(crate) const ACTION_GET_AUTHENTICATED_USER: &str = "GITHUB_GET_THE_AUTHENTICATED_USER"; pub(crate) const ACTION_SEARCH_ISSUES: &str = "GITHUB_SEARCH_ISSUES_AND_PULL_REQUESTS"; -/// Items per search page on steady-state syncs. -const PAGE_SIZE: u32 = 50; - -/// Larger page for the initial post-OAuth backfill. -const INITIAL_PAGE_SIZE: u32 = 100; - -/// Maximum pages per sync pass. Caps initial-backfill churn; the rest rolls -/// over to the next scheduled interval. -const MAX_PAGES: u32 = 20; - const GH_CLI_TIMEOUT: Duration = Duration::from_secs(30); const GITHUB_TASK_SEARCH_TIMEOUT: Duration = Duration::from_secs(20); @@ -124,351 +113,13 @@ impl ComposioProvider for GitHubProvider { }) } + /// Incremental sync via the generic + /// [`orchestrator`](crate::openhuman::memory_sync::composio::providers::orchestrator): + /// login resolution, pagination, dedup, the `max_items` cap, and cursor + /// handling live in `run_sync`; the GitHub-specific primitives — including + /// the **server-side** `sync_depth_days` window — 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:github] incremental sync starting" - ); - - // ── Step 1: load persistent sync state ────────────────────── - let Some(memory) = ctx.memory_client() else { - return Err("[composio:github] memory client not ready".to_string()); - }; - let mut state = SyncState::load(&memory, "github", &connection_id).await?; - - // ── Step 2: check daily budget ─────────────────────────────── - if state.budget_exhausted() { - tracing::info!( - connection_id = %connection_id, - "[composio:github] daily request budget exhausted, skipping sync" - ); - return Ok(SyncOutcome { - toolkit: "github".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: "github sync skipped: daily budget exhausted".to_string(), - details: json!({ "budget_exhausted": true }), - }); - } - - // ── Step 3: resolve the authenticated user's login ────────── - let login = match self.resolve_login(ctx, &mut state).await { - Ok(l) => l, - Err(e) => { - let _ = state.save(&memory).await; - return Err(e); - } - }; - - if state.budget_exhausted() { - tracing::info!( - connection_id = %connection_id, - "[composio:github] budget exhausted after login probe, skipping sync" - ); - state.save(&memory).await?; - return Ok(SyncOutcome { - toolkit: "github".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: "github sync skipped: daily budget exhausted after login probe" - .to_string(), - details: json!({ "budget_exhausted": true, "login_resolved": true }), - }); - } - - // ── Step 4: paginated issue search ─────────────────────────── - // - // `involves:{login}` matches issues/PRs the user created, was assigned - // to, was mentioned in, or commented on — scoped to what GitHub's own - // access rules allow. Combined with `updated:>{cursor}` on subsequent - // runs this converges on a minimal diff 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); - if ctx.max_items.is_some() && effective_max_pages < MAX_PAGES { - tracing::debug!( - connection_id = %connection_id, - max_items = ?ctx.max_items, - effective_max_pages, - "[composio:github] [memory_sync] applying max_items page cap" - ); - } - - // ctx.sync_depth_days: inject `updated:>{date}` on first sync when set. - let depth_query_fragment: Option = if state.cursor.is_none() { - ctx.sync_depth_days.map(|days| { - let floor = chrono::Utc::now() - chrono::Duration::days(days as i64); - let s = floor.format("%Y-%m-%dT%H:%M:%SZ").to_string(); - tracing::debug!( - connection_id = %connection_id, - sync_depth_days = days, - floor = %s, - "[composio:github] [memory_sync] injecting updated:> date filter on first sync" - ); - format!("updated:>{s}") - }) - } else { - None - }; - - // Build the base search query — include depth fragment when set. - let query = build_search_query_with_depth( - &login, - state.cursor.as_deref(), - depth_query_fragment.as_deref(), - ); - - let mut total_fetched: usize = 0; - let mut total_persisted: usize = 0; - let mut newest_updated: Option = None; - // Track whether any per-item ingest failed this pass. If so, we hold - // the persistent cursor — `updated:>{cursor}` on the next search - // would otherwise exclude the failed item, and because the new - // memory-tree pipeline (#2885) is delete-first, an *edited* issue - // 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; - - 'pages: for page_num in 1..=effective_max_pages { - if state.budget_exhausted() { - tracing::info!( - page = page_num, - "[composio:github] budget exhausted mid-sync, stopping pagination" - ); - break; - } - - let args = json!({ - "q": query, - "sort": "updated", - "order": "desc", - "per_page": page_size, - "page": page_num, - }); - - tracing::debug!( - connection_id = %connection_id, - page = page_num, - query = %query, - "[composio:github] executing {ACTION_SEARCH_ISSUES}" - ); - - let resp = match ctx.execute(ACTION_SEARCH_ISSUES, Some(args)).await { - Ok(resp) => resp, - Err(e) => { - let _ = state.save(&memory).await; - return Err(format!( - "[composio:github] {ACTION_SEARCH_ISSUES} 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:github] {ACTION_SEARCH_ISSUES} page={page_num}: {err}" - )); - } - - let issues = sync::extract_issues(&resp.data); - total_fetched += issues.len(); - - if issues.is_empty() { - tracing::debug!( - page = page_num, - "[composio:github] empty page, stopping pagination" - ); - break; - } - - // ── Per-item dedup + persist ───────────────────────────── - for issue in &issues { - let Some(issue_id) = sync::extract_issue_id(issue) else { - tracing::debug!("[composio:github] issue missing id, skipping"); - continue; - }; - - let updated = sync::extract_issue_updated_at(issue); - - // Track the newest `updated_at` for cursor advancement. - if let Some(ref ts) = updated { - if newest_updated.as_ref().is_none_or(|ex| ts > ex) { - newest_updated = Some(ts.clone()); - } - } - - // Composite dedup key: issue_id@updated_at (same trick ClickUp - // uses so that edits after the last sync are re-persisted). - let sync_key = match &updated { - Some(ts) => format!("{issue_id}@{ts}"), - None => issue_id.clone(), - }; - - // If the item's updated_at is at or before our cursor AND we've - // already synced this composite key, every subsequent result on - // this page is guaranteed to be older — stop pagination early. - if let (Some(ref cursor), Some(ref ts)) = (&state.cursor, &updated) { - if ts <= cursor && state.is_synced(&sync_key) { - tracing::debug!( - issue_id = %issue_id, - "[composio:github] reached cursor boundary, stopping" - ); - break 'pages; - } - } - - if state.is_synced(&sync_key) { - continue; - } - - let title_text = sync::extract_issue_title(issue) - .unwrap_or_else(|| format!("GitHub issue {issue_id}")); - - // Route into the memory-tree pipeline (#2885). The prior - // implementation called `persist_single_item` → - // `MemoryClient::store_skill_sync` → UnifiedMemory - // `memory_docs`, which the modern retrieval surfaces - // (`memory.search`, `tree.read_chunk`, `tree.browse`, - // summary trees, MCP tools) don't read from — the data - // was invisible to every agent recall path. - match ingest_issue_into_memory_tree( - &ctx.config, - &connection_id, - &issue_id, - &title_text, - updated.as_deref(), - issue, - ) - .await - { - Ok(_chunks_written) => { - state.mark_synced(&sync_key); - total_persisted += 1; - cap.record(1); - } - Err(e) => { - had_ingest_failures = true; - tracing::warn!( - issue_id = %issue_id, - error = %e, - "[composio:github] failed to ingest issue into memory_tree (continuing)" - ); - } - } - - // ctx.max_items precise cap: stop mid-page so we never persist - // more than the cap even when a single page exceeds it. - if cap.is_reached() { - hit_cap_boundary = true; - break; - } - } - - // ctx.max_items hard stop. - if cap.is_reached() { - tracing::debug!( - page = page_num, - total_persisted, - "[composio:github] [memory_sync] max_items reached, stopping pagination" - ); - hit_cap_boundary = true; - break; - } - - // GitHub search pages are 0-indexed in terms of total results; - // a short page means we've exhausted the result set. - if (issues.len() as u32) < page_size { - tracing::debug!( - page = page_num, - returned = issues.len(), - "[composio:github] short page, end of results" - ); - 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). `set_last_sync_at_ms` - // still advances — that's just a heartbeat, not a fetch-window - // boundary, so it's safe to record that we did attempt a sync. - // Hold the cursor on a cap-truncated pass so the next sync re-scans the unseen tail. - if !had_ingest_failures && !hit_cap_boundary { - if let Some(new_cursor) = newest_updated { - state.advance_cursor(&new_cursor); - } - } else { - tracing::warn!( - connection_id = %connection_id, - had_ingest_failures, - hit_cap_boundary, - "[composio:github] holding cursor — ingest failures or cap-truncated pass; next \ - sync will re-fetch the failed range" - ); - } - state.set_last_sync_at_ms(sync::now_ms()); - state.save(&memory).await?; - - let finished_at_ms = sync::now_ms(); - let summary = format!( - "github 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:github] incremental sync complete" - ); - - Ok(SyncOutcome { - toolkit: "github".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!({ - "issues_fetched": total_fetched, - "issues_persisted": total_persisted, - "budget_remaining": state.budget_remaining(), - "cursor": state.cursor, - "synced_ids_total": state.synced_ids.len(), - }), - }) + run_github_sync(ctx, reason).await } async fn fetch_tasks( @@ -900,43 +551,6 @@ fn extract_github_labels(issue: &serde_json::Value) -> Vec { } } -impl GitHubProvider { - /// Resolve the authenticated user's GitHub login handle. - /// - /// The login is stable for the connection lifetime. We re-fetch on every - /// sync rather than caching in `SyncState` to (a) keep the struct lean - /// and (b) implicitly validate that the OAuth token is still valid before - /// we start paginating search results. - async fn resolve_login( - &self, - ctx: &ProviderContext, - state: &mut SyncState, - ) -> Result { - let resp = ctx - .execute(ACTION_GET_AUTHENTICATED_USER, Some(json!({}))) - .await - .map_err(|e| { - format!("[composio:github] {ACTION_GET_AUTHENTICATED_USER} failed: {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:github] {ACTION_GET_AUTHENTICATED_USER}: {err}" - )); - } - - sync::extract_user_login(&resp.data).ok_or_else(|| { - "[composio:github] GITHUB_GET_THE_AUTHENTICATED_USER returned no login".to_string() - }) - } -} - -/// Build the GitHub Search-Issues query for an incremental sync. /// /// `involves:` is GitHub's logical-OR over `author`, `assignee`, `mentions`, /// and `commenter`, so the result set covers every item the connected user diff --git a/src/openhuman/memory_sync/composio/providers/github/source.rs b/src/openhuman/memory_sync/composio/providers/github/source.rs new file mode 100644 index 000000000..3d3972823 --- /dev/null +++ b/src/openhuman/memory_sync/composio/providers/github/source.rs @@ -0,0 +1,325 @@ +//! GitHub's [`IncrementalSource`] primitives. +//! +//! GitHub rides the generic +//! [`crate::openhuman::memory_sync::composio::providers::orchestrator`]: +//! [`GitHubProvider::sync`](super::provider::GitHubProvider) delegates to +//! [`run_github_sync`]. The orchestrator owns the control flow (budget, +//! pagination bound, dedup, the precise `max_items` clamp, cursor advance/hold, +//! state persistence); this module supplies only the GitHub-specific shapes. +//! +//! GitHub is **flat but identity-scoped**: [`GitHubSource::preamble`] resolves +//! the authenticated login and returns a single [`SyncScope`] carrying it, then +//! the orchestrator pages straight through +//! `GITHUB_SEARCH_ISSUES_AND_PULL_REQUESTS` with `involves:{login}` (1-indexed +//! `page`). Per-item dedup is keyed by `{issue_id}@{updated_at}`. +//! +//! **Server-side depth window.** Unlike the other flat providers, GitHub +//! applies `sync_depth_days` server-side by injecting `updated:>{date}` into the +//! search query on the first sync (before a cursor exists) — so it overrides +//! [`IncrementalSource::server_side_depth`] and the orchestrator skips its +//! client-side timestamp truncation. On incremental syncs the persistent cursor +//! (`updated:>{cursor}`) bounds the window instead. + +use async_trait::async_trait; +use serde_json::{json, Value}; +use std::sync::Mutex; + +use super::provider::{build_search_query_with_depth, ACTION_SEARCH_ISSUES}; +use super::sync; +use crate::openhuman::memory_sync::composio::providers::orchestrator::{ + self, IncrementalSource, IngestOutcome, PageFetch, SyncItem, SyncScope, +}; +use crate::openhuman::memory_sync::composio::providers::sync_state::SyncState; +use crate::openhuman::memory_sync::composio::providers::{ + ProviderContext, SyncOutcome, SyncReason, +}; + +/// Items per search page on steady-state syncs. +const PAGE_SIZE: u32 = 50; + +/// Larger page for the initial post-OAuth backfill. +const INITIAL_PAGE_SIZE: u32 = 100; + +/// Maximum pages per sync pass. +const MAX_PAGES: u32 = 20; + +/// GitHub's [`IncrementalSource`]. +#[derive(Default)] +pub(crate) struct GitHubSource { + depth_fragment: Mutex>, +} + +/// Entry point used by [`super::provider::GitHubProvider::sync`]. +pub(crate) async fn run_github_sync( + ctx: &ProviderContext, + reason: SyncReason, +) -> Result { + orchestrator::run_sync(&GitHubSource::default(), ctx, reason).await +} + +impl GitHubSource { + /// Resolve the authenticated user's GitHub login. Re-fetched every sync + /// (rather than cached in `SyncState`) so it implicitly validates the OAuth + /// token before paginating. Records the request against the budget. + async fn resolve_login( + &self, + ctx: &ProviderContext, + state: &mut SyncState, + ) -> Result { + tracing::debug!( + connection_id = ?ctx.connection_id, + "[composio:github] resolve_login via {}", + super::provider::ACTION_GET_AUTHENTICATED_USER + ); + + let resp = ctx + .execute( + super::provider::ACTION_GET_AUTHENTICATED_USER, + Some(json!({})), + ) + .await + .map_err(|e| { + format!( + "[composio:github] {} failed: {e:#}", + super::provider::ACTION_GET_AUTHENTICATED_USER + ) + })?; + state.record_requests(1); + + if !resp.successful { + let err = resp + .error + .clone() + .unwrap_or_else(|| "provider reported failure".to_string()); + return Err(format!( + "[composio:github] {}: {err}", + super::provider::ACTION_GET_AUTHENTICATED_USER + )); + } + + let login = sync::extract_user_login(&resp.data).ok_or_else(|| { + "[composio:github] GITHUB_GET_THE_AUTHENTICATED_USER returned no login".to_string() + })?; + + tracing::debug!( + connection_id = ?ctx.connection_id, + "[composio:github] resolve_login complete" + ); + Ok(login) + } + + fn set_depth_fragment(&self, fragment: Option) { + let mut guard = self + .depth_fragment + .lock() + .unwrap_or_else(|e| e.into_inner()); + *guard = fragment; + } + + fn depth_fragment(&self) -> Option { + self.depth_fragment + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone() + } +} + +#[async_trait] +impl IncrementalSource for GitHubSource { + fn toolkit(&self) -> &'static str { + "github" + } + + fn page_size(&self, reason: SyncReason) -> u32 { + match reason { + SyncReason::ConnectionCreated => INITIAL_PAGE_SIZE, + _ => PAGE_SIZE, + } + } + + fn max_pages(&self) -> u32 { + MAX_PAGES + } + + /// GitHub applies the depth window server-side (see module docs). + fn server_side_depth(&self) -> bool { + true + } + + fn detail_noun(&self) -> &'static str { + "issues" + } + + /// Resolve the login and carry it as the single scope's id. + async fn preamble( + &self, + ctx: &ProviderContext, + state: &mut SyncState, + ) -> Result, String> { + let depth_fragment = if state.cursor.is_none() { + ctx.sync_depth_days.map(|days| { + let floor = chrono::Utc::now() - chrono::Duration::days(days as i64); + format!("updated:>{}", floor.format("%Y-%m-%dT%H:%M:%SZ")) + }) + } else { + None + }; + self.set_depth_fragment(depth_fragment); + + let login = self.resolve_login(ctx, state).await?; + let label = format!("involves:{login}"); + Ok(vec![SyncScope::nested(login, label)]) + } + + async fn fetch_page( + &self, + ctx: &ProviderContext, + scope: &SyncScope, + cursor: Option<&str>, + reason: SyncReason, + state: &mut SyncState, + ) -> Result { + let login = &scope.id; + // GitHub paginates by 1-indexed `page`; the orchestrator's opaque cursor + // carries the next page number (`None` = first page). + let page_num: u32 = cursor.and_then(|c| c.parse().ok()).unwrap_or(1); + let page_size = self.page_size(reason); + + // Stable for the whole sync pass: GitHub's `page` is relative to the + // exact query, so the depth floor must not move while paginating. + let depth_fragment = self.depth_fragment(); + let query = build_search_query_with_depth( + login, + state.cursor.as_deref(), + depth_fragment.as_deref(), + ); + + let args = json!({ + "q": query, + "sort": "updated", + "order": "desc", + "per_page": page_size, + "page": page_num, + }); + + tracing::debug!( + connection_id = ?ctx.connection_id, + page_num, + page_size, + has_depth_fragment = depth_fragment.is_some(), + "[composio:github] fetch_page via {ACTION_SEARCH_ISSUES}" + ); + + let resp = ctx + .execute(ACTION_SEARCH_ISSUES, Some(args)) + .await + .map_err(|e| { + format!("[composio:github] {ACTION_SEARCH_ISSUES} page={page_num}: {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:github] {ACTION_SEARCH_ISSUES} page={page_num}: {err}" + )); + } + + let issues = sync::extract_issues(&resp.data); + // A short page means the result set is exhausted — no next page. + let next = if (issues.len() as u32) < page_size { + None + } else { + Some((page_num + 1).to_string()) + }; + + tracing::debug!( + connection_id = ?ctx.connection_id, + page_num, + fetched = issues.len(), + has_next = next.is_some(), + "[composio:github] fetch_page complete" + ); + + Ok(PageFetch { + items: issues, + next, + }) + } + + fn item_dedup_key(&self, item: &Value) -> Option { + let issue_id = sync::extract_issue_id(item)?; + match sync::extract_issue_updated_at(item) { + Some(updated) => Some(format!("{issue_id}@{updated}")), + None => Some(issue_id), + } + } + + fn item_sort_ts(&self, item: &Value) -> Option { + sync::extract_issue_updated_at(item) + } + + /// GitHub ingests sequentially (parity with the prior per-item loop) — one + /// issue at a time into the memory-tree pipeline. + 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"); + let mut outcome = IngestOutcome::default(); + for it in items { + let Some(issue_id) = sync::extract_issue_id(&it.raw) else { + continue; + }; + let title = sync::extract_issue_title(&it.raw) + .unwrap_or_else(|| format!("GitHub issue {issue_id}")); + match super::ingest::ingest_issue_into_memory_tree( + &ctx.config, + connection_id, + &issue_id, + &title, + it.sort_ts.as_deref(), + &it.raw, + ) + .await + { + Ok(_chunks_written) => { + outcome.synced_keys.push(it.dedup_key); + outcome.persisted += 1; + } + Err(e) => { + outcome.had_failures = true; + tracing::warn!( + issue_id = %issue_id, + error = %e, + "[composio:github] failed to ingest issue into memory_tree (continuing)" + ); + } + } + } + outcome + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn item_dedup_key_composes_id_and_updated() { + let with_update = json!({ "id": 42, "number": 7, "updated_at": "2026-05-01T00:00:00Z" }); + // GitHub ids may be numeric; extract_issue_id renders them as strings. + let key = GitHubSource::default().item_dedup_key(&with_update); + assert!( + key.as_deref().map(|k| k.contains('@')).unwrap_or(false), + "expected composite id@updated key, got {key:?}" + ); + } +} diff --git a/src/openhuman/memory_sync/composio/providers/linear/mod.rs b/src/openhuman/memory_sync/composio/providers/linear/mod.rs index bb40292b8..532a2892e 100644 --- a/src/openhuman/memory_sync/composio/providers/linear/mod.rs +++ b/src/openhuman/memory_sync/composio/providers/linear/mod.rs @@ -5,6 +5,7 @@ mod ingest; mod provider; +mod source; mod sync; #[cfg(test)] mod tests; diff --git a/src/openhuman/memory_sync/composio/providers/linear/provider.rs b/src/openhuman/memory_sync/composio/providers/linear/provider.rs index 5df1c9a52..e51244117 100644 --- a/src/openhuman/memory_sync/composio/providers/linear/provider.rs +++ b/src/openhuman/memory_sync/composio/providers/linear/provider.rs @@ -19,38 +19,22 @@ //! and avoids accidentally ingesting other teammates' private issues. use async_trait::async_trait; -use futures::StreamExt; -use serde_json::{json, Value}; +use serde_json::json; -use super::{ingest::ingest_issue_into_memory_tree, sync}; -use crate::openhuman::config::Config; -use crate::openhuman::memory_sync::composio::providers::sync_state::{extract_item_id, SyncState}; +use super::source::run_linear_sync; +use super::sync; +use crate::openhuman::memory_sync::composio::providers::sync_state::extract_item_id; use crate::openhuman::memory_sync::composio::providers::{ merge_extra, pick_str, resolve_sync_interval_secs, ComposioProvider, CuratedTool, NormalizedTask, ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason, TaskFetchFilter, TaskKind, }; -const ACTION_LIST_USERS: &str = "LINEAR_LIST_LINEAR_USERS"; -const ACTION_LIST_ISSUES: &str = "LINEAR_LIST_LINEAR_ISSUES"; - -/// Page size per API call. We use a small window on steady-state syncs -/// to keep response sizes bounded. -const PAGE_SIZE: u64 = 50; - -/// Larger initial-sync page size so the first backfill catches up faster. -const INITIAL_PAGE_SIZE: u64 = 100; - -/// Maximum pages per sync pass before yielding. Caps initial backfill -/// churn — anything beyond this rolls over to the next sync interval. -const MAX_PAGES_PER_SYNC: u32 = 20; +pub(super) const ACTION_LIST_USERS: &str = "LINEAR_LIST_LINEAR_USERS"; +pub(super) const ACTION_LIST_ISSUES: &str = "LINEAR_LIST_LINEAR_ISSUES"; /// Paths for extracting a Linear issue's unique ID. -const ISSUE_ID_PATHS: &[&str] = &["id", "data.id", "identifier", "data.identifier"]; - -/// 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; +pub(super) const ISSUE_ID_PATHS: &[&str] = &["id", "data.id", "identifier", "data.identifier"]; pub struct LinearProvider; @@ -126,291 +110,13 @@ impl ComposioProvider for LinearProvider { }) } + /// Incremental sync via the generic + /// [`orchestrator`](crate::openhuman::memory_sync::composio::providers::orchestrator): + /// viewer resolution, pagination, dedup, the `max_items` cap, the + /// `sync_depth_days` window, and cursor handling live in `run_sync`; the + /// Linear-specific primitives 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:linear] incremental sync starting" - ); - - // ── Step 1: load persistent sync state ────────────────────── - let Some(memory) = ctx.memory_client() else { - return Err("[composio:linear] memory client not ready".to_string()); - }; - let mut state = SyncState::load(&memory, "linear", &connection_id).await?; - - // ── Step 2: check daily budget ────────────────────────────── - if state.budget_exhausted() { - tracing::info!( - connection_id = %connection_id, - "[composio:linear] daily request budget exhausted, skipping sync" - ); - return Ok(SyncOutcome { - toolkit: "linear".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: "linear sync skipped: daily budget exhausted".to_string(), - details: json!({ "budget_exhausted": true }), - }); - } - - // ── Step 3: resolve the authenticated user's ID ───────────── - let viewer_id = match self.resolve_viewer_id(ctx, &mut state).await { - Ok(id) => id, - Err(e) => { - let _ = state.save(&memory).await; - return Err(e); - } - }; - - // Re-check budget after the viewer-id probe. - if state.budget_exhausted() { - tracing::info!( - connection_id = %connection_id, - "[composio:linear] budget exhausted after viewer-id probe, skipping sync" - ); - state.save(&memory).await?; - return Ok(SyncOutcome { - toolkit: "linear".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: "linear sync skipped: daily budget exhausted after viewer-id probe" - .to_string(), - details: json!({ "budget_exhausted": true, "viewer_id_resolved": true }), - }); - } - - // ── Step 4: 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 as u32, MAX_PAGES_PER_SYNC); - if ctx.max_items.is_some() && effective_max_pages < MAX_PAGES_PER_SYNC { - tracing::debug!( - connection_id = %connection_id, - max_items = ?ctx.max_items, - effective_max_pages, - "[composio:linear] [memory_sync] applying max_items page cap" - ); - } - - // ctx.sync_depth_days: oldest allowed updatedAt for client-side skip. - let oldest_allowed_time: Option = ctx.sync_depth_days.map(|days| { - let floor = chrono::Utc::now() - chrono::Duration::days(days as i64); - let s = floor.to_rfc3339(); - tracing::debug!( - connection_id = %connection_id, - sync_depth_days = days, - oldest_allowed = %s, - "[composio:linear] [memory_sync] applying sync_depth_days floor" - ); - s - }); - - let mut total_fetched: usize = 0; - let mut total_persisted: usize = 0; - let mut had_persist_failures = false; - let mut newest_updated: Option = None; - let mut after_cursor: Option = None; - let mut hit_cursor_boundary = 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:linear] budget exhausted mid-sync, stopping pagination" - ); - break; - } - - let mut args = json!({ - "assigneeId": &viewer_id, - "first": page_size, - "orderBy": "updatedAt", - }); - - if let Some(ref cursor) = after_cursor { - args["after"] = json!(cursor); - } - - let resp = ctx - .execute(ACTION_LIST_ISSUES, Some(args)) - .await - .map_err(|e| { - format!("[composio:linear] {ACTION_LIST_ISSUES} 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:linear] {ACTION_LIST_ISSUES} page={page_num}: {err}" - )); - } - - let issues = sync::extract_issues(&resp.data); - total_fetched += issues.len(); - - if issues.is_empty() { - tracing::debug!( - page = page_num, - "[composio:linear] empty page, stopping pagination" - ); - break; - } - - // ── Per-item dedup + bounded-concurrency ingest ────────── - let (mut pending, page_hit_boundary) = - select_pending(&issues, &state, &mut newest_updated); - if page_hit_boundary { - hit_cursor_boundary = true; - } - - // ctx.sync_depth_days: drop items updated before the depth floor. `pending` is - // in descending timestamp order, so truncate at the first item below the floor - // and signal cursor-boundary so pagination stops. - if let Some(ref floor) = oldest_allowed_time { - if let Some(cut) = pending.iter().position(|p| { - p.updated - .as_deref() - .map(|t| t < floor.as_str()) - .unwrap_or(false) - }) { - pending.truncate(cut); - hit_cursor_boundary = true; - } - } - - // ctx.max_items: clamp the dedup'd batch to the remaining budget before ingest. - cap.clamp_batch(&mut pending); - - let ingestor = MemoryTreeIngestor { - config: ctx.config.as_ref(), - connection_id: &connection_id, - }; - 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_persist_failures = true; - } - - // ctx.max_items precise cap: once the per-source cap is hit, stop paginating. - if cap.is_reached() { - hit_cap_boundary = true; - break; - } - - if hit_cursor_boundary { - tracing::debug!( - page = page_num, - "[composio:linear] reached cursor boundary, stopping pagination" - ); - break; - } - - // ctx.max_items hard stop. - if cap.is_reached() { - tracing::debug!( - page = page_num, - total_persisted, - "[composio:linear] [memory_sync] max_items reached, stopping pagination" - ); - hit_cap_boundary = true; - break; - } - - // Advance to the next page using Linear's cursor-based pagination. - match sync::extract_pagination_cursor(&resp.data) { - Some(next_cursor) => { - after_cursor = Some(next_cursor); - } - None => { - tracing::debug!( - page = page_num, - "[composio:linear] no next page cursor, end of results" - ); - break; - } - } - } - - // ── Step 5: advance cursor and save state ──────────────────── - // Hold the cursor on a cap-truncated pass so the next sync re-scans the unseen tail. - if had_persist_failures { - tracing::warn!( - "[composio:linear] persist failures seen; keeping previous cursor for retry" - ); - } else if hit_cap_boundary { - tracing::warn!( - hit_cap_boundary, - "[composio:linear] holding cursor — cap-truncated pass; next sync will re-scan \ - the unseen tail" - ); - } else if let Some(new_cursor) = newest_updated { - state.advance_cursor(&new_cursor); - } - state.set_last_sync_at_ms(sync::now_ms()); - state.save(&memory).await?; - - let finished_at_ms = sync::now_ms(); - let summary = format!( - "linear 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:linear] incremental sync complete" - ); - - Ok(SyncOutcome { - toolkit: "linear".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!({ - "issues_fetched": total_fetched, - "issues_persisted": total_persisted, - "budget_remaining": state.budget_remaining(), - "cursor": state.cursor, - "synced_ids_total": state.synced_ids.len(), - }), - }) + run_linear_sync(ctx, reason).await } async fn fetch_tasks( @@ -540,340 +246,3 @@ fn extract_linear_labels(issue: &serde_json::Value) -> Vec { None => Vec::new(), } } - -impl LinearProvider { - /// Look up (and budget-record) the authenticated viewer's ID. - /// - /// The ID is stable for the connection's lifetime. We re-fetch on - /// every sync rather than caching it in `SyncState` because (a) the - /// call is cheap, (b) it implicitly validates that the OAuth - /// connection is still good before we start paginating. - async fn resolve_viewer_id( - &self, - ctx: &ProviderContext, - state: &mut SyncState, - ) -> Result { - let resp = ctx - .execute(ACTION_LIST_USERS, Some(json!({ "isMe": true }))) - .await - .map_err(|e| format!("[composio:linear] {ACTION_LIST_USERS} failed: {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:linear] {ACTION_LIST_USERS}: {err}")); - } - - sync::extract_viewer_id(&resp.data).ok_or_else(|| { - "[composio:linear] LINEAR_LIST_LINEAR_USERS returned no viewer id".to_string() - }) - } -} - -/// One issue that passed dedupe in pass 1 and is queued for concurrent -/// ingest in pass 2. Borrows the raw issue `Value` out of the current -/// page's `issues` (same scope — no clone needed). -struct PendingIngest<'a> { - sync_key: String, - issue_id: String, - title: String, - updated: Option, - issue: &'a Value, -} - -/// 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 issues 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 Linear issue", 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 IssueIngestor { - async fn ingest( - &self, - issue_id: &str, - title: &str, - updated: Option<&str>, - issue: &Value, - ) -> anyhow::Result; -} - -/// Production ingestor: routes into the memory-tree pipeline via -/// [`ingest_issue_into_memory_tree`]. -struct MemoryTreeIngestor<'c> { - config: &'c Config, - connection_id: &'c str, -} - -#[async_trait] -impl IssueIngestor for MemoryTreeIngestor<'_> { - async fn ingest( - &self, - issue_id: &str, - title: &str, - updated: Option<&str>, - issue: &Value, - ) -> anyhow::Result { - ingest_issue_into_memory_tree( - self.config, - self.connection_id, - issue_id, - title, - updated, - issue, - ) - .await - } -} - -/// Pass 1 (pure, no I/O): scan one page of `issues`, advance -/// `newest_updated`, skip already-synced items, and collect the issues -/// 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>( - issues: &'a [Value], - state: &SyncState, - newest_updated: &mut Option, -) -> (Vec>, bool) { - let mut hit_cursor_boundary = false; - let mut pending: Vec = Vec::new(); - for issue in issues { - let Some(issue_id) = extract_item_id(issue, ISSUE_ID_PATHS) else { - tracing::debug!("[composio:linear] issue missing ID, skipping"); - continue; - }; - - let updated = sync::extract_issue_updated(issue); - - // Track newest `updatedAt` for cursor advancement. - if let Some(ref ts) = updated { - if newest_updated.as_ref().is_none_or(|existing| ts > existing) { - *newest_updated = Some(ts.clone()); - } - } - - // Composite (issue_id, updatedAt) key so re-edited issues are - // re-persisted on the next sync. - let sync_key = match &updated { - Some(ts) => format!("{issue_id}@{ts}"), - None => issue_id.clone(), - }; - - // Older than cursor AND already synced → caught up. - if let (Some(ref cursor), Some(ref ts)) = (&state.cursor, &updated) { - if ts <= cursor && state.is_synced(&sync_key) { - hit_cursor_boundary = true; - continue; - } - } - - if state.is_synced(&sync_key) { - continue; - } - - let title_text = - sync::extract_issue_title(issue).unwrap_or_else(|| format!("Linear issue {issue_id}")); - let title = format!("Linear: {title_text}"); - - pending.push(PendingIngest { - sync_key, - issue_id, - title, - updated, - issue, - }); - } - (pending, hit_cursor_boundary) -} - -/// Pass 2: ingest the queued issues 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`. A failed ingest is logged and skipped, tripping -/// `had_failures` so the caller holds the cursor (parity with the -/// previous sequential path). -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.issue_id, &p.title, p.updated.as_deref(), p.issue) - .await; - (p.sync_key, p.issue_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, issue_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!( - issue_id = %issue_id, - error = %e, - "[composio:linear] failed to ingest issue into memory_tree (continuing)" - ); - } - } - } - outcome -} - -#[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 `issue_id`. No - /// memory tree or embedder involved — lets us assert the concurrency - /// bound and overlap deterministically. - struct CountingIngestor { - in_flight: AtomicUsize, - peak: AtomicUsize, - fail_issue: Option, - } - - impl CountingIngestor { - fn new(fail_issue: Option<&str>) -> Arc { - Arc::new(Self { - in_flight: AtomicUsize::new(0), - peak: AtomicUsize::new(0), - fail_issue: fail_issue.map(str::to_string), - }) - } - } - - #[async_trait] - impl IssueIngestor for CountingIngestor { - async fn ingest( - &self, - issue_id: &str, - _title: &str, - _updated: Option<&str>, - _issue: &Value, - ) -> 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_issue.as_deref() == Some(issue_id) { - Err(anyhow::anyhow!("forced failure for {issue_id}")) - } else { - Ok(1) - } - } - } - - fn make_issues(n: usize) -> Vec { - (0..n).map(|i| json!({ "id": format!("i{i}") })).collect() - } - - fn make_pending(issues: &[Value]) -> Vec> { - issues - .iter() - .enumerate() - .map(|(i, issue)| PendingIngest { - sync_key: format!("k{i}"), - issue_id: format!("i{i}"), - title: format!("Linear: issue {i}"), - updated: None, - issue, - }) - .collect() - } - - #[tokio::test] - async fn ingest_pending_buffered_bounds_and_overlaps() { - let ingestor = CountingIngestor::new(None); - let issues = make_issues(20); - let pending = make_pending(&issues); - - let outcome = ingest_pending_buffered(ingestor.as_ref(), pending, 8).await; - - assert_eq!(outcome.persisted, 20, "all issues 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("i2")); - let issues = make_issues(5); - let pending = make_pending(&issues); - - 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 issue's sync_key must not be marked synced" - ); - } - - #[test] - fn select_pending_tracks_newest_skips_synced_and_detects_boundary() { - let mut state = SyncState::new("linear", "conn1"); - state.cursor = Some("2026-04-15T00:00:00Z".to_string()); - // Issue B is already synced and older than the cursor. - state.mark_synced("b@2026-04-01T00:00:00Z"); - - let issues = vec![ - json!({ "id": "a", "updatedAt": "2026-05-01T00:00:00Z" }), - json!({ "id": "b", "updatedAt": "2026-04-01T00:00:00Z" }), - json!({ "updatedAt": "2026-03-01T00:00:00Z" }), // no id → skipped - ]; - - let mut newest: Option = None; - let (pending, hit_boundary) = select_pending(&issues, &state, &mut newest); - - assert_eq!(pending.len(), 1, "only the new issue A is queued"); - assert_eq!(pending[0].issue_id, "a"); - assert_eq!(pending[0].sync_key, "a@2026-05-01T00:00:00Z"); - assert!( - hit_boundary, - "older synced issue 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/linear/source.rs b/src/openhuman/memory_sync/composio/providers/linear/source.rs new file mode 100644 index 000000000..84343e7f3 --- /dev/null +++ b/src/openhuman/memory_sync/composio/providers/linear/source.rs @@ -0,0 +1,397 @@ +//! Linear's [`IncrementalSource`] primitives. +//! +//! Linear rides the generic +//! [`crate::openhuman::memory_sync::composio::providers::orchestrator`]: +//! [`LinearProvider::sync`](super::provider::LinearProvider) delegates to +//! [`run_linear_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 Linear-specific shapes. +//! +//! Linear is **flat but identity-scoped**: [`LinearSource::preamble`] resolves +//! the viewer id and returns a single [`SyncScope`] carrying it, then the +//! orchestrator pages straight through `LINEAR_LIST_LINEAR_ISSUES` filtered to +//! that assignee. Pagination is GraphQL cursor based (`after` / `endCursor`); +//! the depth window is applied client-side (RFC3339 `updatedAt`). Per-item +//! dedup is keyed by `{issue_id}@{updatedAt}` so an edited issue re-ingests. + +use async_trait::async_trait; +use futures::StreamExt; +use serde_json::{json, Value}; + +use super::provider::{ACTION_LIST_ISSUES, ACTION_LIST_USERS, ISSUE_ID_PATHS}; +use super::{ingest::ingest_issue_into_memory_tree, 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, +}; + +/// Page size per API call on steady-state syncs. +const PAGE_SIZE: u32 = 50; + +/// Larger initial-sync page size so the first backfill catches up faster. +const INITIAL_PAGE_SIZE: u32 = 100; + +/// Maximum pages per sync pass before yielding. +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; + +/// Linear's [`IncrementalSource`]. +pub(crate) struct LinearSource; + +/// Entry point used by [`super::provider::LinearProvider::sync`]. +pub(crate) async fn run_linear_sync( + ctx: &ProviderContext, + reason: SyncReason, +) -> Result { + orchestrator::run_sync(&LinearSource, ctx, reason).await +} + +#[async_trait] +impl IncrementalSource for LinearSource { + fn toolkit(&self) -> &'static str { + "linear" + } + + 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 + } + + fn detail_noun(&self) -> &'static str { + "issues" + } + + /// Resolve the viewer id and return it as the single scope's id. + async fn preamble( + &self, + ctx: &ProviderContext, + state: &mut SyncState, + ) -> Result, String> { + let resp = ctx + .execute(ACTION_LIST_USERS, Some(json!({ "isMe": true }))) + .await + .map_err(|e| format!("[composio:linear] {ACTION_LIST_USERS} failed: {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:linear] {ACTION_LIST_USERS}: {err}")); + } + + let viewer_id = sync::extract_viewer_id(&resp.data).ok_or_else(|| { + "[composio:linear] LINEAR_LIST_LINEAR_USERS returned no viewer id".to_string() + })?; + Ok(vec![SyncScope::nested(viewer_id, "assignee:me")]) + } + + async fn fetch_page( + &self, + ctx: &ProviderContext, + scope: &SyncScope, + cursor: Option<&str>, + reason: SyncReason, + state: &mut SyncState, + ) -> Result { + let mut args = json!({ + "assigneeId": &scope.id, + "first": self.page_size(reason), + "orderBy": "updatedAt", + }); + if let Some(cursor) = cursor { + args["after"] = json!(cursor); + } + + let resp = ctx + .execute(ACTION_LIST_ISSUES, Some(args)) + .await + .map_err(|e| format!("[composio:linear] {ACTION_LIST_ISSUES}: {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:linear] {ACTION_LIST_ISSUES}: {err}")); + } + + Ok(PageFetch { + items: sync::extract_issues(&resp.data), + next: sync::extract_pagination_cursor(&resp.data), + }) + } + + fn item_dedup_key(&self, item: &Value) -> Option { + let issue_id = extract_item_id(item, ISSUE_ID_PATHS)?; + match sync::extract_issue_updated(item) { + Some(updated) => Some(format!("{issue_id}@{updated}")), + None => Some(issue_id), + } + } + + fn item_sort_ts(&self, item: &Value) -> Option { + sync::extract_issue_updated(item) + } + + 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"); + + let pending: Vec = items + .into_iter() + .filter_map(|it| { + let issue_id = extract_item_id(&it.raw, ISSUE_ID_PATHS)?; + let title_text = sync::extract_issue_title(&it.raw) + .unwrap_or_else(|| format!("Linear issue {issue_id}")); + Some(PendingIngest { + sync_key: it.dedup_key, + issue_id, + title: format!("Linear: {title_text}"), + updated: it.sort_ts, + issue: it.raw, + }) + }) + .collect(); + + 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 issue queued for concurrent ingest. Owns its raw issue `Value` (the +/// orchestrator handed ownership via [`SyncItem`]). +struct PendingIngest { + sync_key: String, + issue_id: String, + title: String, + updated: Option, + issue: Value, +} + +/// Folded result of [`ingest_pending_buffered`]. Order-independent. +#[derive(Default)] +struct BufferedIngestOutcome { + synced_keys: Vec, + persisted: usize, + had_failures: bool, +} + +/// Seam over "ingest one Linear issue", so the bounded-concurrency driver can +/// be unit-tested with a fake that records peak in-flight calls. +#[async_trait] +trait IssueIngestor { + async fn ingest( + &self, + issue_id: &str, + title: &str, + updated: Option<&str>, + issue: &Value, + ) -> anyhow::Result; +} + +/// Production ingestor: routes into the memory-tree pipeline. +struct MemoryTreeIngestor<'c> { + config: &'c Config, + connection_id: &'c str, +} + +#[async_trait] +impl IssueIngestor for MemoryTreeIngestor<'_> { + async fn ingest( + &self, + issue_id: &str, + title: &str, + updated: Option<&str>, + issue: &Value, + ) -> anyhow::Result { + ingest_issue_into_memory_tree( + self.config, + self.connection_id, + issue_id, + title, + updated, + issue, + ) + .await + } +} + +/// Ingest queued issues with bounded concurrency, folding into an +/// order-independent [`BufferedIngestOutcome`]. A failed ingest is logged and +/// skipped, tripping `had_failures` so the orchestrator holds the cursor. +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.issue_id, &p.title, p.updated.as_deref(), &p.issue) + .await; + (p.sync_key, p.issue_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, issue_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!( + issue_id = %issue_id, + error = %e, + "[composio:linear] failed to ingest issue into memory_tree (continuing)" + ); + } + } + } + outcome +} + +#[cfg(test)] +mod buffered_tests { + use super::*; + use serde_json::json; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + /// Fake ingestor: records peak concurrent in-flight `ingest` calls and can + /// fail one specific `issue_id`. No memory tree or embedder involved. + struct CountingIngestor { + in_flight: AtomicUsize, + peak: AtomicUsize, + fail_issue: Option, + } + + impl CountingIngestor { + fn new(fail_issue: Option<&str>) -> Arc { + Arc::new(Self { + in_flight: AtomicUsize::new(0), + peak: AtomicUsize::new(0), + fail_issue: fail_issue.map(str::to_string), + }) + } + } + + #[async_trait] + impl IssueIngestor for CountingIngestor { + async fn ingest( + &self, + issue_id: &str, + _title: &str, + _updated: Option<&str>, + _issue: &Value, + ) -> 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_issue.as_deref() == Some(issue_id) { + Err(anyhow::anyhow!("forced failure for {issue_id}")) + } else { + Ok(1) + } + } + } + + fn make_pending(n: usize) -> Vec { + (0..n) + .map(|i| PendingIngest { + sync_key: format!("k{i}"), + issue_id: format!("i{i}"), + title: format!("Linear: issue {i}"), + updated: None, + issue: json!({ "id": format!("i{i}") }), + }) + .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 issues 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("i2")); + 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 issue's sync_key must not be marked synced" + ); + } + + #[test] + fn item_dedup_key_composes_id_and_updated() { + let with_update = json!({ "id": "i1", "updatedAt": "2026-05-01T00:00:00Z" }); + assert_eq!( + LinearSource.item_dedup_key(&with_update).as_deref(), + Some("i1@2026-05-01T00:00:00Z") + ); + let no_update = json!({ "id": "i2" }); + assert_eq!( + LinearSource.item_dedup_key(&no_update).as_deref(), + Some("i2") + ); + assert_eq!( + LinearSource.item_dedup_key(&json!({ "updatedAt": "x" })), + None + ); + } +} diff --git a/src/openhuman/memory_sync/composio/providers/orchestrator.rs b/src/openhuman/memory_sync/composio/providers/orchestrator.rs index ad56a2da3..324163673 100644 --- a/src/openhuman/memory_sync/composio/providers/orchestrator.rs +++ b/src/openhuman/memory_sync/composio/providers/orchestrator.rs @@ -173,12 +173,33 @@ pub(crate) trait IncrementalSource: Send + Sync { /// 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. + /// (clickup) override. Unused when [`Self::server_side_depth`] is `true`. fn depth_floor(&self, days: u32) -> String { let floor = chrono::Utc::now() - chrono::Duration::days(days as i64); floor.to_rfc3339() } + /// Noun used for this provider's `{noun}_fetched` / `{noun}_persisted` + /// keys in the [`SyncOutcome::details`] diagnostic blob, preserving each + /// provider's historical detail shape (notion: `results`, github/linear: + /// `issues`, clickup: `tasks`, …). `details` is for logging/UI status only; + /// nothing reads these keys in production. + fn detail_noun(&self) -> &'static str { + "results" + } + + /// Whether the provider applies the `sync_depth_days` window **itself** + /// (server-side — e.g. GitHub's `updated:>{date}` search qualifier), + /// rather than relying on the orchestrator's client-side timestamp + /// truncation. When `true`, the orchestrator skips its client-side depth + /// filter and the provider must inject the window inside + /// [`Self::fetch_page`] (typically only on the first sync, before a cursor + /// exists). Default `false` — the orchestrator filters client-side via + /// [`Self::depth_floor`]. + fn server_side_depth(&self) -> bool { + false + } + /// 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 @@ -376,17 +397,23 @@ pub(crate) async fn run_sync( ); } - 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 - }); + // Server-side-depth providers (GitHub) inject the window into the request + // in `fetch_page`, so the orchestrator skips its client-side floor for them. + let depth_floor: Option = if source.server_side_depth() { + None + } else { + 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; @@ -540,6 +567,20 @@ pub(crate) async fn run_sync( "[composio:sync_orch] incremental sync complete" ); + // Provider-named `{noun}_fetched` / `{noun}_persisted` keys preserve each + // provider's historical `details` shape (notion `results`, github/linear + // `issues`, …). Built dynamically since `json!` can't take runtime keys. + let noun = source.detail_noun(); + let mut details = json!({ + "budget_remaining": state.budget_remaining(), + "cursor": state.cursor, + "synced_ids_total": state.synced_ids.len(), + }); + if let Some(obj) = details.as_object_mut() { + obj.insert(format!("{noun}_fetched"), json!(total_fetched)); + obj.insert(format!("{noun}_persisted"), json!(total_persisted)); + } + Ok(SyncOutcome { toolkit: toolkit.to_string(), connection_id: Some(connection_id), @@ -548,13 +589,7 @@ pub(crate) async fn run_sync( 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(), - }), + details, }) } @@ -586,6 +621,9 @@ mod tests { /// provider-reported failure — pins that a failed page still consumes /// the daily budget. provider_fail_fetch: bool, + /// When true, advertise server-side depth so the orchestrator skips its + /// client-side window filter (GitHub's behaviour). + server_side_depth: bool, } impl FakeSource { @@ -669,6 +707,9 @@ mod tests { fn item_sort_ts(&self, item: &Value) -> Option { item.get("ts").and_then(Value::as_str).map(str::to_string) } + fn server_side_depth(&self) -> bool { + self.server_side_depth + } async fn ingest( &self, _ctx: &ProviderContext, @@ -761,6 +802,33 @@ mod tests { ); } + #[tokio::test] + async fn server_side_depth_skips_the_client_side_filter() { + // Same ancient items, but the source advertises server-side depth — so + // the orchestrator must NOT client-side-truncate (the provider would + // have filtered in fetch_page). All five survive here. + let tmp = TempDir::new().unwrap(); + let ctx = fake_ctx(&tmp, None, Some(7)); + let items = vec![ + json!({ "id": "a", "ts": "2099-01-02T00:00:00Z" }), + json!({ "id": "b", "ts": "2000-01-01T00:00:00Z" }), + json!({ "id": "c", "ts": "2000-01-02T00:00:00Z" }), + ]; + let source = FakeSource { + scopes: vec![SyncScope::flat()], + explicit_items: Some(items), + server_side_depth: true, + ..Default::default() + }; + let outcome = run_sync(&source, &ctx, SyncReason::Manual) + .await + .expect("run_sync"); + assert_eq!( + outcome.items_ingested, 3, + "server_side_depth must skip the orchestrator's client-side window filter" + ); + } + #[tokio::test] async fn nested_scopes_share_one_cap_budget() { let tmp = TempDir::new().unwrap(); diff --git a/tests/memory_sync_providers_raw_coverage_e2e.rs b/tests/memory_sync_providers_raw_coverage_e2e.rs index 2f8a4bb54..13e9bc061 100644 --- a/tests/memory_sync_providers_raw_coverage_e2e.rs +++ b/tests/memory_sync_providers_raw_coverage_e2e.rs @@ -1293,6 +1293,86 @@ async fn linear_sync_max_items_caps_ingest_to_exact_count() { server.abort(); } +// ───────────────────────────────────────────────────────────────────────── +// Linear sync_depth_days enforcement (via the shared orchestrator) +// +// Linear applies the depth window client-side (RFC3339 `updatedAt`). The mock +// returns 2 recent issues (far-future) + 3 ancient (year-2000) in descending +// order; with sync_depth_days=7 only the 2 recent issues persist — the +// orchestrator truncates the page at the first item below the floor. +// ───────────────────────────────────────────────────────────────────────── + +/// Build `recent` + `old` Linear issues in descending `updatedAt` order. +fn linear_depth_issues(recent: usize, old: usize) -> Vec { + let mut issues = Vec::new(); + for i in 0..recent { + issues.push(json!({ + "id": format!("linear-recent-{i:04}"), + "identifier": format!("ENG-R{i}"), + "title": format!("Recent {i}"), + "updatedAt": format!("2099-12-{:02}T10:00:00.000Z", 28 - i), + "url": format!("https://linear.app/cap/issue/ENG-R{i}"), + "description": "recent", + })); + } + for i in 0..old { + issues.push(json!({ + "id": format!("linear-old-{i:04}"), + "identifier": format!("ENG-O{i}"), + "title": format!("Old {i}"), + "updatedAt": format!("2000-01-{:02}T10:00:00.000Z", 3 - i), + "url": format!("https://linear.app/cap/issue/ENG-O{i}"), + "description": "old", + })); + } + issues +} + +#[tokio::test] +async fn linear_sync_depth_days_filters_old_issues() { + 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 + // issues must be ingested. + let requests: Arc>> = Arc::new(Mutex::new(Vec::new())); + let (base, server) = loopback_router(linear_cap_router( + linear_depth_issues(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: "linear".to_string(), + connection_id: Some("conn-linear-depth".to_string()), + usage: Default::default(), + max_items: None, + sync_depth_days: Some(7), + }; + + let outcome = LinearProvider::new() + .sync(&ctx, SyncReason::ConnectionCreated) + .await + .expect("linear depth sync"); + + assert_eq!( + outcome.items_ingested, 2, + "sync_depth_days=7 must drop the 3 year-2000 issues and keep the 2 recent ones" + ); + + server.abort(); +} + // ───────────────────────────────────────────────────────────────────────── // ClickUp cap enforcement //