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): convert GitHub + Linear to the generic sync orchestrator (Refs #3333, stacked on #3852) (#3860)

Browse Source 
Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: Steven Enamakel <enamakel@tinyhumans.ai>
This commit is contained in:
sanil-23
2026-06-21 15:20:30 -07:00
committed by GitHub
co-authored by Claude Steven Enamakel
parent 6d220a7b2a
commit b4a2242266
8 changed files with 911 additions and 1056 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openhuman/memory_sync/composio/providers/github/mod.rs
+1
View File
@@ -14,6 +14,7 @@
mod ingest;
mod provider;
mod source;
mod sync;
#[cfg(test)]
mod tests;
src/openhuman/memory_sync/composio/providers/github/provider.rs
+7 -393
View File
@@ -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<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: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<String> = if state.cursor.is_none() {
ctx.sync_depth_days.map(|days| {
let floor = chrono::Utc::now() - chrono::Duration::days(days as i64);
let s = floor.format("%Y-%m-%dT%H:%M:%SZ").to_string();
tracing::debug!(
connection_id = %connection_id,
sync_depth_days = days,
floor = %s,
"[composio:github] [memory_sync] injecting updated:> date filter on first sync"
);
format!("updated:>{s}")
})
} else {
None
};
// Build the base search query — include depth fragment when set.
let query = build_search_query_with_depth(
&login,
state.cursor.as_deref(),
depth_query_fragment.as_deref(),
);
let mut total_fetched: usize = 0;
let mut total_persisted: usize = 0;
let mut newest_updated: Option<String> = 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<String> {
}
}
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<String, String> {
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
src/openhuman/memory_sync/composio/providers/github/source.rs
+325
View File
@@ -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<Option<String>>,
}
/// Entry point used by [`super::provider::GitHubProvider::sync`].
pub(crate) async fn run_github_sync(
ctx: &ProviderContext,
reason: SyncReason,
) -> Result<SyncOutcome, String> {
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<String, String> {
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<String>) {
let mut guard = self
.depth_fragment
.lock()
.unwrap_or_else(|e| e.into_inner());
*guard = fragment;
}
fn depth_fragment(&self) -> Option<String> {
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<Vec<SyncScope>, 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<PageFetch, String> {
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<String> {
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<String> {
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<SyncItem>,
) -> 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:?}"
);
}
}
src/openhuman/memory_sync/composio/providers/linear/mod.rs
+1
View File
@@ -5,6 +5,7 @@
mod ingest;
mod provider;
mod source;
mod sync;
#[cfg(test)]
mod tests;
src/openhuman/memory_sync/composio/providers/linear/provider.rs
+13 -644
View File
@@ -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<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: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<String> = ctx.sync_depth_days.map(|days| {
let floor = chrono::Utc::now() - chrono::Duration::days(days as i64);
let s = floor.to_rfc3339();
tracing::debug!(
connection_id = %connection_id,
sync_depth_days = days,
oldest_allowed = %s,
"[composio:linear] [memory_sync] applying sync_depth_days floor"
);
s
});
let mut total_fetched: usize = 0;
let mut total_persisted: usize = 0;
let mut had_persist_failures = false;
let mut newest_updated: Option<String> = None;
let mut after_cursor: Option<String> = None;
let mut hit_cursor_boundary = false;
let mut hit_cap_boundary = false;
for page_num in 0..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<String> {
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<String, 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}"));
}
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<String>,
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<String>,
/// 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<usize>;
}
/// 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<usize> {
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<String>,
) -> (Vec<PendingIngest<'a>>, bool) {
let mut hit_cursor_boundary = false;
let mut pending: Vec<PendingIngest> = 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<I: IssueIngestor + 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.issue_id, &p.title, p.updated.as_deref(), p.issue)
.await;
(p.sync_key, p.issue_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, 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<String>,
}
impl CountingIngestor {
fn new(fail_issue: Option<&str>) -> Arc<Self> {
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<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_issue.as_deref() == Some(issue_id) {
Err(anyhow::anyhow!("forced failure for {issue_id}"))
} else {
Ok(1)
}
}
}
fn make_issues(n: usize) -> Vec<Value> {
(0..n).map(|i| json!({ "id": format!("i{i}") })).collect()
}
fn make_pending(issues: &[Value]) -> Vec<PendingIngest<'_>> {
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<String> = 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"));
}
}
src/openhuman/memory_sync/composio/providers/linear/source.rs
+397
View File
@@ -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<SyncOutcome, String> {
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<Vec<SyncScope>, 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<PageFetch, String> {
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<String> {
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<String> {
sync::extract_issue_updated(item)
}
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");
let pending: Vec<PendingIngest> = 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<String>,
issue: Value,
}
/// Folded result of [`ingest_pending_buffered`]. Order-independent.
#[derive(Default)]
struct BufferedIngestOutcome {
synced_keys: Vec<String>,
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<usize>;
}
/// 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<usize> {
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<I: IssueIngestor + Sync>(
ingestor: &I,
pending: Vec<PendingIngest>,
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::<Vec<_>>();
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<String>,
}
impl CountingIngestor {
fn new(fail_issue: Option<&str>) -> Arc<Self> {
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<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_issue.as_deref() == Some(issue_id) {
Err(anyhow::anyhow!("forced failure for {issue_id}"))
} else {
Ok(1)
}
}
}
fn make_pending(n: usize) -> Vec<PendingIngest> {
(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
);
}
}
src/openhuman/memory_sync/composio/providers/orchestrator.rs
+87 -19
View File
@@ -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<S: IncrementalSource + ?Sized>(
);
}
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
});
// 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<String> = 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<S: IncrementalSource + ?Sized>(
"[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<S: IncrementalSource + ?Sized>(
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<String> {
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();
tests/memory_sync_providers_raw_coverage_e2e.rs
+80
View File
@@ -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<Value> {
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<Mutex<Vec<Value>>> = 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
//