From e95c892dba1d634050e6cbbd8a702533c52cca73 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Mon, 22 Jun 2026 22:39:34 +0530 Subject: [PATCH] refactor(memory_sync): convert Slack to the generic sync orchestrator (#3924) --- .../composio/providers/orchestrator.rs | 331 ++++++++++++-- .../composio/providers/slack/mod.rs | 1 + .../composio/providers/slack/provider.rs | 366 +-------------- .../composio/providers/slack/source.rs | 422 ++++++++++++++++++ .../memory_sync_slack_bus_raw_coverage_e2e.rs | 5 +- 5 files changed, 741 insertions(+), 384 deletions(-) create mode 100644 src/openhuman/memory_sync/composio/providers/slack/source.rs diff --git a/src/openhuman/memory_sync/composio/providers/orchestrator.rs b/src/openhuman/memory_sync/composio/providers/orchestrator.rs index 324163673..79521493b 100644 --- a/src/openhuman/memory_sync/composio/providers/orchestrator.rs +++ b/src/openhuman/memory_sync/composio/providers/orchestrator.rs @@ -19,17 +19,22 @@ //! [`SyncScope`] is the abstraction that lets the same orchestrator drive both //! shapes: //! -//! * **Flat** providers (gmail, github, notion, linear) expose a *single -//! implicit scope* — [`SyncScope::flat`] — and page straight through their -//! one result stream. +//! * **Flat** providers (github, notion, linear) expose a *single implicit +//! scope* — [`SyncScope::flat`] — and page straight through their one +//! result stream. //! * **Nested** providers (clickup workspaces → tasks, slack channels → //! history) resolve their containers in [`IncrementalSource::preamble`] and //! return one [`SyncScope`] per container; the orchestrator's //! `for scope { for page {…} }` loop is byte-for-byte identical for both. //! -//! Only Notion (flat) rides this path today, but the scope loop is written so -//! the nested providers slot in without a control-flow change — see the -//! migration checklist in the issue. +//! ## Per-scope cursors + error tolerance (Slack) +//! +//! Slack is the structural outlier: it keeps a watermark **per channel** (not a +//! single global cursor) and must not let one bad channel abort the rest. Two +//! opt-in trait hooks — [`IncrementalSource::per_scope_cursors`] and +//! [`IncrementalSource::tolerate_scope_errors`] — bend the same scope loop to +//! that shape without touching the single-cursor providers, which leave both +//! `false` and keep the advance-once-at-the-end path verbatim. use async_trait::async_trait; use serde_json::{json, Value}; @@ -209,6 +214,44 @@ pub(crate) trait IncrementalSource: Send + Sync { true } + /// Whether this source keeps a **cursor per scope** (Slack: one `oldest` + /// watermark per channel) instead of the single global watermark every + /// other provider uses. Default `false`. + /// + /// When `true` the orchestrator: + /// * disables its global-cursor boundary detection in [`select_pending`] + /// (the scope's watermark is read by the provider's own + /// [`Self::fetch_page`] — usually injected as a server-side `oldest` + /// filter, so `server_side_depth` is typically `true` too); + /// * tracks the newest timestamp **per scope** and, after each scope + /// finishes cleanly, calls [`Self::advance_scope_cursor`] to advance + /// that one scope's watermark, persisting state between scopes; + /// * holds a scope's watermark (does not advance it) when that scope hit + /// an ingest failure or the `max_items` cap truncated it — so the next + /// pass re-fetches that scope's unseen/failed tail. + /// + /// Single-cursor providers leave this `false` and keep the global + /// advance-once-at-the-end path verbatim. + fn per_scope_cursors(&self) -> bool { + false + } + + /// Whether a scope-level failure is **non-fatal**. Default `false` — a + /// `fetch_page` error aborts the whole pass (every single-cursor provider's + /// behaviour). When `true` (Slack), a `fetch_page` error or a scope's + /// ingest failure is logged, counted in `scopes_errored`, and the + /// orchestrator continues to the next scope instead of returning `Err`. + fn tolerate_scope_errors(&self) -> bool { + false + } + + /// Advance the persisted watermark for a single `scope` to `newest_ts` + /// (per-scope-cursor mode only). The default is a no-op; Slack overrides it + /// to fold the new watermark into its per-channel cursor map serialized in + /// [`SyncState::cursor`]. Never called when [`Self::per_scope_cursors`] is + /// `false`. + fn advance_scope_cursor(&self, _state: &mut SyncState, _scope: &SyncScope, _newest_ts: &str) {} + /// Persist a batch of already-filtered items. May spend budget via `state` /// (e.g. Notion's per-page body fetch). Returns which dedup keys succeeded /// so the orchestrator can mark them synced. @@ -260,9 +303,15 @@ fn skipped_outcome( /// This is the generic form of every provider's old `select_pending`. All /// order-dependent decisions live here so the (possibly concurrent) ingest /// stage never has to reason about ordering. +/// +/// `boundary_cursor` is the watermark used for cursor-boundary detection. +/// Single-cursor providers pass `state.cursor`; per-scope-cursor providers +/// (Slack) pass `None` to disable the global-cursor boundary entirely — their +/// per-scope watermark is enforced server-side inside `fetch_page` instead. fn select_pending( source: &S, items: &[Value], + boundary_cursor: Option<&str>, state: &SyncState, newest_ts: &mut Option, ) -> (Vec, bool) { @@ -288,8 +337,8 @@ fn select_pending( } // Older-or-equal to the cursor AND already synced → we have caught up. - if let (Some(cursor), Some(ts)) = (&state.cursor, &sort_ts) { - if ts <= cursor && state.is_synced(&dedup_key) { + if let (Some(cursor), Some(ts)) = (boundary_cursor, &sort_ts) { + if ts.as_str() <= cursor && state.is_synced(&dedup_key) { hit_cursor_boundary = true; continue; } @@ -416,15 +465,31 @@ pub(crate) async fn run_sync( }; // ── Step 5: scope × page loop ─────────────────────────────────────── + // + // `per_scope` providers (Slack) advance a watermark per scope inside the + // loop and hold it on a cap-truncated / failed scope; single-cursor + // providers advance one global watermark once, in Step 6. `tolerate` + // providers continue past a scope-level failure instead of aborting. + let per_scope = source.per_scope_cursors(); + let tolerate = source.tolerate_scope_errors(); + let mut total_fetched: usize = 0; let mut total_persisted: usize = 0; let mut newest_ts: Option = None; let mut had_ingest_failures = false; let mut hit_cap_boundary = false; + let mut scopes_synced: usize = 0; + let mut scopes_errored: usize = 0; 'scopes: for scope in &scopes { // The page cursor is per-scope — reset at the top of every scope. let mut cursor: Option = None; + // Newest timestamp observed *within this scope*, for per-scope advance. + let mut scope_newest_ts: Option = None; + // This scope hit an ingest failure (hold its watermark, count errored). + let mut scope_had_failure = false; + // This scope was truncated by the global cap (hold its watermark). + let mut scope_hit_cap = false; for page_num in 0..effective_max_pages { if state.budget_exhausted() { @@ -441,13 +506,28 @@ pub(crate) async fn run_sync( // provider-reported failures) per its contract. On error we persist // whatever budget/dedup progress we have before propagating — // parity with the per-provider loops, which saved state before - // returning a failed-page error. + // returning a failed-page error. When the source tolerates + // scope-level errors (Slack), we instead log, count, persist, and + // move on to the next scope. let fetch = match source .fetch_page(ctx, scope, cursor.as_deref(), reason, &mut state) .await { Ok(fetch) => fetch, Err(e) => { + if tolerate { + tracing::warn!( + toolkit, + connection_id = %connection_id, + scope = %scope.label, + page = page_num, + error = %e, + "[composio:sync_orch] scope fetch failed (continuing with next scope)" + ); + scopes_errored += 1; + let _ = state.save(&memory).await; + continue 'scopes; + } let _ = state.save(&memory).await; return Err(e); } @@ -464,9 +544,21 @@ pub(crate) async fn run_sync( break; } - // Dedup + cursor-boundary detection + newest-ts tracking. + // Dedup + cursor-boundary detection + newest-ts tracking. Per-scope + // providers disable the global-cursor boundary (pass `None`) and + // accumulate the newest ts into their per-scope tracker. + let boundary_cursor = if per_scope { + None + } else { + state.cursor.as_deref() + }; + let ts_acc = if per_scope { + &mut scope_newest_ts + } else { + &mut newest_ts + }; let (mut pending, mut hit_cursor_boundary) = - select_pending(source, &fetch.items, &state, &mut newest_ts); + select_pending(source, &fetch.items, boundary_cursor, &state, ts_acc); // sync_depth_days: `pending` is in descending-timestamp order, so // truncate at the first item below the floor and stop paginating. @@ -495,12 +587,15 @@ pub(crate) async fn run_sync( cap.record(outcome.persisted); if outcome.had_failures { had_ingest_failures = true; + scope_had_failure = true; } - // Precise cap reached → stop the entire pass. + // Precise cap reached → stop the entire pass (after settling this + // scope's watermark below). if cap.is_reached() { hit_cap_boundary = true; - break 'scopes; + scope_hit_cap = true; + break; } if hit_cursor_boundary { @@ -524,28 +619,72 @@ pub(crate) async fn run_sync( break; } } + + // ── Per-scope watermark settle (per_scope providers only) ──────── + if per_scope { + if scope_had_failure { + // Ingest failure → hold this scope's watermark, count errored. + scopes_errored += 1; + tracing::warn!( + toolkit, + connection_id = %connection_id, + scope = %scope.label, + "[composio:sync_orch] scope ingest failed; watermark held, re-fetch next pass" + ); + } else { + if scope_hit_cap { + // Cap truncated this scope → hold its watermark so the next + // pass re-scans the unseen tail; still a processed scope. + tracing::debug!( + toolkit, + scope = %scope.label, + "[composio:sync_orch] cap truncated scope; watermark held" + ); + } else if let Some(ref nt) = scope_newest_ts { + source.advance_scope_cursor(&mut state, scope, nt); + } + scopes_synced += 1; + } + // Persist between scopes for crash-resilience (parity with the + // per-provider Slack loop, which saved state after each channel). + if let Err(err) = state.save(&memory).await { + tracing::warn!( + toolkit, + error = %err, + "[composio:sync_orch] per-scope state save failed (non-fatal)" + ); + } + } + + if scope_hit_cap { + break 'scopes; + } } // ── Step 6: advance cursor (or hold) and persist state ────────────── // - // Hold the cursor on a cap-truncated pass so the next sync re-scans the + // Per-scope providers already advanced/held each scope's watermark inline, + // so the global advance is skipped for them. Single-cursor providers hold + // the global cursor on a cap-truncated pass so the next sync re-scans the // unseen tail, and on an ingest failure when the source opts in (Notion's // delete-first safety). Otherwise advance to the newest timestamp seen. - let hold_cursor = - hit_cap_boundary || (had_ingest_failures && source.hold_cursor_on_ingest_failure()); - if !hold_cursor { - if let Some(new_cursor) = newest_ts { - state.advance_cursor(&new_cursor); + if !per_scope { + let hold_cursor = + hit_cap_boundary || (had_ingest_failures && source.hold_cursor_on_ingest_failure()); + if !hold_cursor { + if let Some(new_cursor) = newest_ts { + state.advance_cursor(&new_cursor); + } + } else { + tracing::warn!( + toolkit, + connection_id = %connection_id, + had_ingest_failures, + hit_cap_boundary, + "[composio:sync_orch] holding cursor — cap-truncated pass or ingest failures; \ + next sync will re-fetch the unseen/failed range" + ); } - } else { - tracing::warn!( - toolkit, - connection_id = %connection_id, - had_ingest_failures, - hit_cap_boundary, - "[composio:sync_orch] holding cursor — cap-truncated pass or ingest failures; \ - next sync will re-fetch the unseen/failed range" - ); } state.set_last_sync_at_ms(now_ms()); state.save(&memory).await?; @@ -579,6 +718,14 @@ pub(crate) async fn run_sync( if let Some(obj) = details.as_object_mut() { obj.insert(format!("{noun}_fetched"), json!(total_fetched)); obj.insert(format!("{noun}_persisted"), json!(total_persisted)); + // Scope accounting is only meaningful for providers that iterate real + // per-scope work with tolerance/advance (Slack); emitting it for + // single-cursor providers would change their historical `details` shape. + if per_scope || tolerate { + obj.insert("scopes_total".to_string(), json!(scopes.len())); + obj.insert("scopes_synced".to_string(), json!(scopes_synced)); + obj.insert("scopes_errored".to_string(), json!(scopes_errored)); + } } Ok(SyncOutcome { @@ -624,6 +771,15 @@ mod tests { /// When true, advertise server-side depth so the orchestrator skips its /// client-side window filter (GitHub's behaviour). server_side_depth: bool, + /// When true, keep a per-scope watermark map in `state.cursor` + /// (Slack's behaviour) instead of a single global cursor. + per_scope: bool, + /// When true, a scope-level fetch failure is non-fatal — the pass + /// continues to the next scope and records `scopes_errored`. + tolerate: bool, + /// Scope id whose `fetch_page` returns a transport error (used to + /// exercise per-scope error tolerance). `None` → every scope succeeds. + fail_scope: Option, } impl FakeSource { @@ -659,6 +815,23 @@ mod tests { } Ok(self.scopes.clone()) } + fn per_scope_cursors(&self) -> bool { + self.per_scope + } + fn tolerate_scope_errors(&self) -> bool { + self.tolerate + } + fn advance_scope_cursor(&self, state: &mut SyncState, scope: &SyncScope, newest_ts: &str) { + // Mirror Slack: fold the watermark into a per-scope map serialized + // in `state.cursor`. + let mut map: std::collections::BTreeMap = state + .cursor + .as_deref() + .and_then(|s| serde_json::from_str(s).ok()) + .unwrap_or_default(); + map.insert(scope.id.clone(), newest_ts.to_string()); + state.cursor = Some(serde_json::to_string(&map).unwrap()); + } async fn fetch_page( &self, _ctx: &ProviderContext, @@ -667,6 +840,10 @@ mod tests { _reason: SyncReason, state: &mut SyncState, ) -> Result { + if self.fail_scope.as_deref() == Some(scope.id.as_str()) { + // A scope-level transport error — exercises tolerate_scope_errors. + return Err(format!("fake scope fetch failure for {}", scope.id)); + } if self.fail_fetch { // Simulate a transport error (no completed round-trip) → not // recorded, matching the contract. @@ -982,7 +1159,13 @@ mod tests { ]; let mut newest: Option = None; - let (pending, hit_boundary) = select_pending(&source, &items, &state, &mut newest); + let (pending, hit_boundary) = select_pending( + &source, + &items, + state.cursor.as_deref(), + &state, + &mut newest, + ); assert_eq!(pending.len(), 1, "only the new item A survives"); assert_eq!(pending[0].dedup_key, "a"); @@ -992,4 +1175,94 @@ mod tests { ); assert_eq!(newest.as_deref(), Some("2026-05-01T00:00:00Z")); } + + #[tokio::test] + async fn per_scope_cursors_advance_each_scope_independently() { + let tmp = TempDir::new().unwrap(); + let ctx = fake_ctx(&tmp, None, None); + let source = FakeSource { + scopes: vec![ + SyncScope::nested("s1", "Scope 1"), + SyncScope::nested("s2", "Scope 2"), + ], + items_per_scope: 3, + per_scope: true, + ..Default::default() + }; + let outcome = run_sync(&source, &ctx, SyncReason::Periodic) + .await + .expect("run_sync"); + assert_eq!(outcome.items_ingested, 6, "both scopes' items ingested"); + assert_eq!(outcome.details["scopes_synced"], 2); + assert_eq!(outcome.details["scopes_errored"], 0); + + // The persisted cursor is a per-scope watermark map carrying *both* + // scopes — proof the orchestrator advanced each scope independently. + let memory = ctx.memory_client().expect("memory client"); + let state = SyncState::load(&memory, "faketoolkit", "conn-fake") + .await + .unwrap(); + let map: std::collections::BTreeMap = + serde_json::from_str(state.cursor.as_deref().unwrap()).expect("cursor map"); + assert_eq!(map.len(), 2, "both scopes have a watermark"); + assert!(map.contains_key("s1") && map.contains_key("s2")); + } + + #[tokio::test] + async fn tolerate_scope_errors_continues_past_a_failed_scope() { + let tmp = TempDir::new().unwrap(); + let ctx = fake_ctx(&tmp, None, None); + // Scope s1 fails its fetch; s2 succeeds. With tolerance the pass + // ingests s2's items, counts s1 errored, and never returns Err. + let source = FakeSource { + scopes: vec![ + SyncScope::nested("s1", "Scope 1"), + SyncScope::nested("s2", "Scope 2"), + ], + items_per_scope: 3, + per_scope: true, + tolerate: true, + fail_scope: Some("s1".to_string()), + ..Default::default() + }; + let outcome = run_sync(&source, &ctx, SyncReason::Periodic) + .await + .expect("tolerant run_sync must not error"); + assert_eq!( + outcome.items_ingested, 3, + "only the healthy scope's items are ingested" + ); + assert_eq!(outcome.details["scopes_errored"], 1); + assert_eq!(outcome.details["scopes_synced"], 1); + + // The failed scope advanced no watermark; only the healthy one did. + let memory = ctx.memory_client().expect("memory client"); + let state = SyncState::load(&memory, "faketoolkit", "conn-fake") + .await + .unwrap(); + let map: std::collections::BTreeMap = + serde_json::from_str(state.cursor.as_deref().unwrap_or("{}")).expect("cursor map"); + assert!(map.contains_key("s2"), "healthy scope advanced"); + assert!(!map.contains_key("s1"), "failed scope watermark held"); + } + + #[tokio::test] + async fn per_scope_fatal_when_tolerance_off() { + // Same failing scope, but tolerance off → the fetch error aborts the + // whole pass (single-cursor providers' default behaviour). + let tmp = TempDir::new().unwrap(); + let ctx = fake_ctx(&tmp, None, None); + let source = FakeSource { + scopes: vec![SyncScope::nested("s1", "Scope 1")], + items_per_scope: 3, + per_scope: true, + tolerate: false, + fail_scope: Some("s1".to_string()), + ..Default::default() + }; + let err = run_sync(&source, &ctx, SyncReason::Periodic) + .await + .expect_err("without tolerance a scope fetch error propagates"); + assert!(err.contains("scope fetch failure"), "got: {err}"); + } } diff --git a/src/openhuman/memory_sync/composio/providers/slack/mod.rs b/src/openhuman/memory_sync/composio/providers/slack/mod.rs index ab9d98f63..31c6b29ed 100644 --- a/src/openhuman/memory_sync/composio/providers/slack/mod.rs +++ b/src/openhuman/memory_sync/composio/providers/slack/mod.rs @@ -17,6 +17,7 @@ pub mod types; pub mod users; mod provider; +mod source; pub use provider::{run_backfill_via_search, SlackProvider, BACKFILL_DAYS}; pub use schemas::{all_slack_memory_controller_schemas, all_slack_memory_registered_controllers}; diff --git a/src/openhuman/memory_sync/composio/providers/slack/provider.rs b/src/openhuman/memory_sync/composio/providers/slack/provider.rs index 6cdd8dea6..dea477e10 100644 --- a/src/openhuman/memory_sync/composio/providers/slack/provider.rs +++ b/src/openhuman/memory_sync/composio/providers/slack/provider.rs @@ -54,7 +54,7 @@ use crate::openhuman::memory_sync::composio::providers::{ /// Composio action slug for channel listing. const ACTION_LIST_CONVERSATIONS: &str = "SLACK_LIST_CONVERSATIONS"; /// Composio action slug for message history. -const ACTION_FETCH_HISTORY: &str = "SLACK_FETCH_CONVERSATION_HISTORY"; +pub(super) const ACTION_FETCH_HISTORY: &str = "SLACK_FETCH_CONVERSATION_HISTORY"; /// Composio action slug for team/workspace profile fetch. const ACTION_FETCH_TEAM_INFO: &str = "SLACK_FETCH_TEAM_INFO"; /// Composio action slug for Slack `auth.test` — returns the authed @@ -72,7 +72,7 @@ pub const BACKFILL_DAYS: i64 = 6; /// Resolve the active backfill window in days. Reads /// `OPENHUMAN_SLACK_BACKFILL_DAYS` env var if set and parseable as a /// positive integer; falls back to [`BACKFILL_DAYS`] otherwise. -fn backfill_days() -> i64 { +pub(super) fn backfill_days() -> i64 { match std::env::var("OPENHUMAN_SLACK_BACKFILL_DAYS") { Ok(s) => match s.trim().parse::() { Ok(n) if n >= 1 => n, @@ -92,10 +92,10 @@ fn backfill_days() -> i64 { const LIST_PAGE_SIZE: u32 = 200; /// Max messages per `SLACK_FETCH_CONVERSATION_HISTORY` page. -const HISTORY_PAGE_SIZE: u32 = 1000; +pub(super) const HISTORY_PAGE_SIZE: u32 = 1000; /// Stop paginating any single channel's history after this many pages. -const MAX_HISTORY_PAGES_PER_CHANNEL: u32 = 20; +pub(super) const MAX_HISTORY_PAGES_PER_CHANNEL: u32 = 20; /// Stop paginating channel listings after this many pages. const MAX_LIST_PAGES: u32 = 10; @@ -405,179 +405,14 @@ impl ComposioProvider for SlackProvider { Ok(profile) } + /// Slack rides the generic orchestrator. Channel enumeration + the user + /// directory backfill happen in [`super::source::SlackSource::preamble`]; + /// per-channel `conversations.history` pagination, the per-channel `oldest` + /// watermark, dedup, the `max_items` cap, and per-channel error tolerance + /// all live in `run_sync`. The Slack-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:slack] sync starting" - ); - - let Some(memory) = ctx.memory_client() else { - return Err("[composio:slack] memory client not ready".to_string()); - }; - let mut state = SyncState::load(&memory, "slack", &connection_id).await?; - - if state.budget_exhausted() { - tracing::info!( - connection_id = %connection_id, - "[composio:slack] daily request budget exhausted, skipping sync" - ); - return Ok(SyncOutcome { - toolkit: "slack".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: "slack sync skipped: daily budget exhausted".to_string(), - details: json!({ "budget_exhausted": true }), - }); - } - - let mut cursors = sync::decode_cursors(state.cursor.as_deref()); - let now = chrono::Utc::now(); - - // Pull the workspace user directory once per sync. - let (users, user_call_count) = SlackUsers::fetch(ctx).await; - state.record_requests(user_call_count); - tracing::info!( - connection_id = %connection_id, - user_count = users.len(), - "[composio:slack] users cached for this sync" - ); - - // 1. Enumerate channels. - let channels = list_all_channels(ctx, &mut state) - .await - .map_err(|e| format!("[composio:slack] list_channels: {e:#}"))?; - - tracing::info!( - connection_id = %connection_id, - channel_count = channels.len(), - "[composio:slack] channels discovered" - ); - - let _ = state.save(&memory).await; - - let mut total_messages_ingested: usize = 0; - let mut channels_processed: usize = 0; - let mut channels_errored: usize = 0; - let mut hit_cap_boundary = false; - - // ctx.max_items: ItemCap is threaded through process_channel so the - // per-page batch is clamped before ingest and the channel loop stops - // precisely at the cap — the old coarse post-channel check allowed a - // single page/channel to blow past the cap. - let mut cap = super::super::helpers::ItemCap::new(ctx.max_items); - - // 2. Per-channel: fetch → post-process → enrich → ingest. - for channel in &channels { - if state.budget_exhausted() { - tracing::warn!( - connection_id = %connection_id, - channel = %channel.id, - "[composio:slack] budget exhausted mid-sync, remaining channels deferred" - ); - break; - } - - match process_channel( - ctx, - &mut state, - channel, - &mut cursors, - now, - &users, - &connection_id, - &mut cap, - ) - .await - { - Ok(n) => { - total_messages_ingested += n; - channels_processed += 1; - } - Err(err) => { - channels_errored += 1; - tracing::warn!( - connection_id = %connection_id, - channel = %channel.id, - error = %err, - "[composio:slack] channel sync failed (continuing with next channel)" - ); - } - } - - // ctx.max_items hard stop across all channels (precise — cap was - // already applied inside process_channel so this break fires - // exactly when the budget is exhausted, not one channel later). - if cap.is_reached() { - // Hold the cursor on a cap-truncated pass so the next sync re-scans the unseen tail. - hit_cap_boundary = true; - tracing::debug!( - connection_id = %connection_id, - total_messages_ingested, - "[composio:slack] [memory_sync] max_items reached, stopping channel iteration" - ); - // Save state before breaking without advancing the cursor. - if let Err(err) = state.save(&memory).await { - tracing::warn!( - error = %err, - "[composio:slack] state save failed after cap-stop (non-fatal)" - ); - } - break; - } - - state.advance_cursor(sync::encode_cursors(&cursors)); - if let Err(err) = state.save(&memory).await { - tracing::warn!( - error = %err, - "[composio:slack] state save failed after channel (non-fatal)" - ); - } - } - - if hit_cap_boundary { - // Hold the cursor on a cap-truncated pass so the next sync re-scans the unseen tail. - tracing::warn!( - connection_id = %connection_id, - "[composio:slack] cap-truncated pass; cursor held so next sync re-scans the \ - unseen tail" - ); - } - - let finished_at_ms = sync::now_ms(); - let summary = format!( - "slack sync: channels_processed={channels_processed} \ - channels_errored={channels_errored} \ - messages_ingested={total_messages_ingested}" - ); - tracing::info!( - connection_id = %connection_id, - elapsed_ms = finished_at_ms.saturating_sub(started_at_ms), - "{summary}" - ); - - Ok(SyncOutcome { - toolkit: "slack".to_string(), - connection_id: Some(connection_id), - reason: reason.as_str().to_string(), - items_ingested: total_messages_ingested, - started_at_ms, - finished_at_ms, - summary, - details: json!({ - "channels_processed": channels_processed, - "channels_errored": channels_errored, - }), - }) + super::source::run_slack_sync(ctx, reason).await } async fn on_trigger( @@ -600,7 +435,7 @@ impl ComposioProvider for SlackProvider { /// Paginate through `SLACK_LIST_CONVERSATIONS` and flatten into a /// single `Vec`. -async fn list_all_channels( +pub(super) async fn list_all_channels( ctx: &ProviderContext, state: &mut SyncState, ) -> Result, String> { @@ -645,183 +480,6 @@ async fn list_all_channels( Ok(out) } -/// Pull one channel's history since its cursor, post-process + enrich each -/// page, then ingest all messages. Returns the number of messages written. -/// -/// `cap` is the shared [`super::super::helpers::ItemCap`] for the sync pass. -/// Each page's message batch is clamped to the remaining budget before ingest -/// so the per-sync `max_items` limit is respected precisely regardless of how -/// many messages a single channel/page returns. -async fn process_channel( - ctx: &ProviderContext, - state: &mut SyncState, - channel: &SlackChannel, - cursors: &mut sync::ChannelCursors, - now: chrono::DateTime, - users: &SlackUsers, - connection_id: &str, - cap: &mut super::super::helpers::ItemCap, -) -> Result { - // Cursor value is a raw Slack `ts` (`"."`) preserved - // with full precision, so multi-message-per-second channels don't - // replay the whole second on the next incremental fetch. When no - // cursor exists yet, fall back to `.000000`. - // ctx.sync_depth_days wins over the env-var OPENHUMAN_SLACK_BACKFILL_DAYS - // default when set — it comes from the user-configured source entry. - let oldest_ts = cursors.get(&channel.id).cloned().unwrap_or_else(|| { - let depth_days = ctx - .sync_depth_days - .map(|d| d as i64) - .unwrap_or_else(backfill_days); - let secs = (now - chrono::Duration::days(depth_days)).timestamp(); - tracing::debug!( - channel = %channel.id, - depth_days, - oldest_ts_secs = secs, - "[composio:slack] [memory_sync] computing oldest_ts for backfill" - ); - format!("{secs}.000000") - }); - - let mut all_messages: Vec = Vec::new(); - let mut cursor: Option = None; - - for page_num in 0..MAX_HISTORY_PAGES_PER_CHANNEL { - if state.budget_exhausted() { - tracing::warn!( - channel = %channel.id, - page = page_num, - "[composio:slack] budget exhausted during history fetch" - ); - break; - } - - let mut args = json!({ - "channel": channel.id, - "oldest": oldest_ts.clone(), - "inclusive": false, - "limit": HISTORY_PAGE_SIZE, - }); - if let Some(ref c) = cursor { - args["cursor"] = json!(c); - } - - let (mut resp, attempts) = execute_with_retry( - ctx, - ACTION_FETCH_HISTORY, - args, - &format!( - "{ACTION_FETCH_HISTORY} channel={} page {page_num}", - channel.id - ), - ) - .await?; - state.record_requests(attempts); - dump_response(&channel.id, "history", page_num, &resp.data); - - // Post-process to slim envelope, then enrich with channel context + users. - super::post_process::post_process(ACTION_FETCH_HISTORY, None, &mut resp.data); - let msgs = sync::extract_messages(&resp.data, channel, users); - tracing::debug!( - channel = %channel.id, - page = page_num, - fetched = msgs.len(), - "[composio:slack] history page" - ); - if msgs.is_empty() { - break; - } - all_messages.extend(msgs); - - // Stop fetching further pages for this channel if we have already - // accumulated enough to fill the remaining budget (checked against - // remaining() which accounts for items recorded by previous channels). - if let Some(remaining) = cap.remaining() { - if all_messages.len() >= remaining { - tracing::debug!( - channel = %channel.id, - page = page_num, - accumulated = all_messages.len(), - remaining, - "[composio:slack] [memory_sync] budget nearly full, stopping history pagination" - ); - break; - } - } - - cursor = sync::extract_next_cursor(&resp.data); - if cursor.is_none() { - break; - } - } - - if all_messages.is_empty() { - tracing::debug!( - channel = %channel.id, - "[composio:slack] no new messages" - ); - return Ok(0); - } - - // ctx.max_items precise cap: clamp the full accumulated batch to the - // remaining budget before ingest so we never persist more than the cap - // allows, even if a single channel/page returned more than what remains. - cap.clamp_batch(&mut all_messages); - - if all_messages.is_empty() { - tracing::debug!( - channel = %channel.id, - "[composio:slack] [memory_sync] cap already reached, skipping channel ingest" - ); - return Ok(0); - } - - let msg_count = all_messages.len(); - tracing::info!( - channel = %channel.id, - messages = msg_count, - "[composio:slack] ingesting channel messages" - ); - - match ingest_page_into_memory_tree(&ctx.config, "", connection_id, &all_messages).await { - Ok(chunks) => { - // Advance cursor to the raw `ts` of the latest successfully- - // ingested message. We pick "latest" by the parsed - // (seconds, micros) tuple — lexicographic sort on the raw - // string would also work for the common 10-digit-seconds - // workspace, but the explicit numeric compare is robust to - // the rare older/wider format and skips the load-bearing - // assumption. - if let Some(latest) = all_messages - .iter() - .max_by_key(|m| sync::parse_ts_components(&m.ts_raw)) - .map(|m| m.ts_raw.clone()) - { - cursors.insert(channel.id.clone(), latest); - } - cap.record(msg_count); - tracing::info!( - channel = %channel.id, - messages = msg_count, - chunks, - "[composio:slack] channel ingest done" - ); - // Return message count (consistent with the sync path which - // counts messages, not chunks, for the items_ingested metric). - Ok(msg_count) - } - Err(e) => { - tracing::warn!( - channel = %channel.id, - error = %e, - "[composio:slack] ingest_page_into_memory_tree failed (cursor not advanced)" - ); - // Don't advance cursor — next sync re-fetches this range. - Err(format!("ingest failed for channel {}: {e:#}", channel.id)) - } - } -} - // ── Search-based backfill (one-shot) ──────────────────────────────── /// Composio action slug for workspace-wide message search. diff --git a/src/openhuman/memory_sync/composio/providers/slack/source.rs b/src/openhuman/memory_sync/composio/providers/slack/source.rs new file mode 100644 index 000000000..2a698de97 --- /dev/null +++ b/src/openhuman/memory_sync/composio/providers/slack/source.rs @@ -0,0 +1,422 @@ +//! Slack's [`IncrementalSource`] primitives. +//! +//! Slack rides the generic +//! [`crate::openhuman::memory_sync::composio::providers::orchestrator`]: +//! [`SlackProvider::sync`](super::provider::SlackProvider) delegates to +//! [`run_slack_sync`]. The orchestrator owns the control flow (budget, +//! pagination bound, dedup, the `max_items` clamp, cursor advance/hold, state +//! persistence); this module supplies only the Slack-specific shapes. +//! +//! Slack is the **structural outlier** the orchestrator's per-scope extensions +//! were built for: +//! +//! * **Per-scope cursors** — Slack keeps one `oldest` watermark *per channel* +//! (a `BTreeMap` serialized into [`SyncState::cursor`]) +//! rather than a single global watermark. [`SlackSource`] opts in via +//! [`IncrementalSource::per_scope_cursors`] and advances each channel's +//! watermark in [`IncrementalSource::advance_scope_cursor`]. The watermark +//! is enforced **server-side** (the `oldest` request arg), so the source +//! also advertises [`IncrementalSource::server_side_depth`] and the +//! orchestrator skips its client-side depth/boundary filtering. +//! * **Per-scope error tolerance** — a single channel that errors mid-sync +//! must not abort the other channels, so [`SlackSource`] opts into +//! [`IncrementalSource::tolerate_scope_errors`]. +//! +//! Slack dedups purely by content-hashed chunk ids (UPSERT) plus the per-channel +//! `oldest` watermark, so [`SlackSource::ingest`] returns **no** `synced_keys` +//! — the `synced_ids` set deliberately stays empty (it would otherwise grow +//! unboundedly, one entry per message, forever). +//! +//! The user directory is fetched once as a [`IncrementalSource::preamble`] +//! side-effect and stashed for every per-channel [`IncrementalSource::ingest`] +//! to resolve authors + `<@…>` mentions. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::sync::OnceLock; + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; +use serde_json::{json, Value}; + +use super::ingest::ingest_page_into_memory_tree; +use super::provider::{ + backfill_days, dump_response, execute_with_retry, list_all_channels, ACTION_FETCH_HISTORY, + HISTORY_PAGE_SIZE, MAX_HISTORY_PAGES_PER_CHANNEL, +}; +use super::sync; +use super::types::SlackChannel; +use super::users::SlackUsers; +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, +}; + +/// Slack's [`IncrementalSource`]. Holds the per-sync user directory and channel +/// metadata resolved in the preamble so every per-channel `fetch_page` / +/// `ingest` can read them back. +pub(crate) struct SlackSource { + /// Workspace user directory, fetched once in [`Self::preamble`]. + users: OnceLock, + /// Channel metadata keyed by channel id, resolved in the preamble so + /// `ingest` can render channel labels without re-listing. + channels: OnceLock>, + /// Wall-clock captured once per sync, used to compute the first-fetch + /// backfill window for channels with no cursor yet. + now: DateTime, + /// Monotonic counter for `dump_response` filenames (best-effort debug). + dump_seq: AtomicU32, +} + +impl SlackSource { + fn new() -> Self { + Self { + users: OnceLock::new(), + channels: OnceLock::new(), + now: Utc::now(), + dump_seq: AtomicU32::new(0), + } + } +} + +/// Entry point used by [`super::provider::SlackProvider::sync`]. +pub(crate) async fn run_slack_sync( + ctx: &ProviderContext, + reason: SyncReason, +) -> Result { + orchestrator::run_sync(&SlackSource::new(), ctx, reason).await +} + +#[async_trait] +impl IncrementalSource for SlackSource { + fn toolkit(&self) -> &'static str { + "slack" + } + + fn page_size(&self, _reason: SyncReason) -> u32 { + HISTORY_PAGE_SIZE + } + + fn max_pages(&self) -> u32 { + MAX_HISTORY_PAGES_PER_CHANNEL + } + + fn detail_noun(&self) -> &'static str { + "messages" + } + + /// Slack keeps a watermark per channel, enforced server-side via `oldest`. + fn per_scope_cursors(&self) -> bool { + true + } + + /// A single channel's failure must not abort the rest of the sync. + fn tolerate_scope_errors(&self) -> bool { + true + } + + /// `fetch_page` injects the `oldest` window itself, so the orchestrator must + /// not also apply a client-side depth floor (Slack `ts` isn't RFC3339). + fn server_side_depth(&self) -> bool { + true + } + + /// Resolve the workspace user directory + the channel list, stash both for + /// `ingest`, and return one [`SyncScope`] per channel. + async fn preamble( + &self, + ctx: &ProviderContext, + state: &mut SyncState, + ) -> Result, String> { + // Pull the workspace user directory once per sync (soft-fails to empty). + let (users, user_call_count) = SlackUsers::fetch(ctx).await; + state.record_requests(user_call_count); + tracing::info!( + connection_id = ?ctx.connection_id, + user_count = users.len(), + "[composio:slack] users cached for this sync" + ); + let _ = self.users.set(users); + + // Enumerate channels (records its own page budget). + let channels = list_all_channels(ctx, state) + .await + .map_err(|e| format!("[composio:slack] list_channels: {e:#}"))?; + tracing::info!( + connection_id = ?ctx.connection_id, + channel_count = channels.len(), + "[composio:slack] channels discovered" + ); + + let scopes: Vec = channels + .iter() + .map(|c| { + let label = super::ingest::channel_label(&c.name, c.is_private); + SyncScope::nested(c.id.clone(), label) + }) + .collect(); + + let channel_map: HashMap = + channels.into_iter().map(|c| (c.id.clone(), c)).collect(); + let _ = self.channels.set(channel_map); + + Ok(scopes) + } + + async fn fetch_page( + &self, + ctx: &ProviderContext, + scope: &SyncScope, + cursor: Option<&str>, + _reason: SyncReason, + state: &mut SyncState, + ) -> Result { + let channel_id = &scope.id; + + // Per-channel `oldest` watermark: the persisted cursor for this channel, + // or the backfill window when the channel has never been synced. Full + // microsecond precision is preserved so `inclusive=false` excludes only + // the exact last-seen message. `ctx.sync_depth_days` (user-configured) + // wins over the env-var default when set. + let cursors = sync::decode_cursors(state.cursor.as_deref()); + let oldest_ts = cursors.get(channel_id).cloned().unwrap_or_else(|| { + let depth_days = ctx + .sync_depth_days + .map(|d| d as i64) + .unwrap_or_else(backfill_days); + let secs = (self.now - chrono::Duration::days(depth_days)).timestamp(); + tracing::debug!( + channel = %channel_id, + depth_days, + oldest_ts_secs = secs, + "[composio:slack] [memory_sync] computing oldest_ts for backfill" + ); + format!("{secs}.000000") + }); + + let mut args = json!({ + "channel": channel_id, + "oldest": oldest_ts, + "inclusive": false, + "limit": HISTORY_PAGE_SIZE, + }); + if let Some(c) = cursor { + args["cursor"] = json!(c); + } + + let (mut resp, attempts) = execute_with_retry( + ctx, + ACTION_FETCH_HISTORY, + args, + &format!("{ACTION_FETCH_HISTORY} channel={channel_id}"), + ) + .await?; + state.record_requests(attempts); + + let idx = self.dump_seq.fetch_add(1, Ordering::Relaxed); + dump_response(channel_id, "history", idx, &resp.data); + + // Post-process to the slim envelope, then hand the orchestrator the raw + // message values — `item_dedup_key` drops blanks, `ingest` enriches. + super::post_process::post_process(ACTION_FETCH_HISTORY, None, &mut resp.data); + let items = resp + .data + .get("messages") + .and_then(Value::as_array) + .cloned() + .unwrap_or_default(); + let next = sync::extract_next_cursor(&resp.data); + + tracing::debug!( + channel = %channel_id, + fetched = items.len(), + has_next = next.is_some(), + "[composio:slack] history page" + ); + + Ok(PageFetch { items, next }) + } + + /// Dedup key doubles as the validity gate: a message with no parseable `ts` + /// or blank text returns `None` so the orchestrator drops it **before** the + /// `max_items` clamp — keeping the cap counted against *valid* messages, + /// exactly as the old per-channel loop did (it filtered then clamped). + fn item_dedup_key(&self, item: &Value) -> Option { + let ts = item.get("ts").and_then(Value::as_str)?; + sync::parse_ts(ts)?; + let text = item.get("text").and_then(Value::as_str).unwrap_or(""); + if text.trim().is_empty() { + return None; + } + Some(ts.to_string()) + } + + fn item_sort_ts(&self, item: &Value) -> Option { + item.get("ts").and_then(Value::as_str).map(str::to_string) + } + + /// Advance this channel's watermark inside the per-channel cursor map + /// serialized in [`SyncState::cursor`]. + fn advance_scope_cursor(&self, state: &mut SyncState, scope: &SyncScope, newest_ts: &str) { + let mut cursors = sync::decode_cursors(state.cursor.as_deref()); + cursors.insert(scope.id.clone(), newest_ts.to_string()); + state.cursor = Some(sync::encode_cursors(&cursors)); + } + + async fn ingest( + &self, + ctx: &ProviderContext, + scope: &SyncScope, + _state: &mut SyncState, + items: Vec, + ) -> IngestOutcome { + let channel_id = &scope.id; + // Channel metadata from the preamble (falls back to a bare id-named + // channel if the map is somehow missing this scope). + let channel = self + .channels + .get() + .and_then(|m| m.get(channel_id)) + .cloned() + .unwrap_or_else(|| SlackChannel { + id: channel_id.clone(), + name: channel_id.clone(), + is_private: false, + }); + let users = self.users.get().cloned().unwrap_or_else(SlackUsers::empty); + + // Enrich the surviving raw message values into canonical SlackMessages + // (author resolution + `<@…>` mention rewriting) via the shared parser. + let raws: Vec = items.into_iter().map(|it| it.raw).collect(); + let wrapped = json!({ "messages": raws }); + let messages = sync::extract_messages(&wrapped, &channel, &users); + + if messages.is_empty() { + return IngestOutcome::default(); + } + + let connection_id = ctx.connection_id.as_deref().unwrap_or("default"); + let count = messages.len(); + match ingest_page_into_memory_tree(&ctx.config, "", connection_id, &messages).await { + Ok(chunks) => { + tracing::info!( + channel = %channel_id, + messages = count, + chunks, + "[composio:slack] channel ingest done" + ); + IngestOutcome { + // No synced_keys: Slack dedups via content-hash UPSERT + the + // per-channel watermark, so `synced_ids` stays empty. + synced_keys: Vec::new(), + persisted: count, + had_failures: false, + } + } + Err(e) => { + tracing::warn!( + channel = %channel_id, + error = %e, + "[composio:slack] ingest_page_into_memory_tree failed (watermark held)" + ); + IngestOutcome { + synced_keys: Vec::new(), + persisted: 0, + had_failures: true, + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn source() -> SlackSource { + SlackSource::new() + } + + #[test] + fn item_dedup_key_drops_blank_and_unparseable() { + let s = source(); + // Valid message → keyed by ts. + assert_eq!( + s.item_dedup_key(&json!({ "ts": "1714003200.000100", "text": "hi" })) + .as_deref(), + Some("1714003200.000100") + ); + // Blank text → dropped (None), so it never counts against the cap. + assert_eq!( + s.item_dedup_key(&json!({ "ts": "1714003200.000100", "text": " " })), + None + ); + // Bot-authored (no `user`) but non-blank text → kept (parity with the + // old extractor, which retained author-less messages). + assert_eq!( + s.item_dedup_key(&json!({ "ts": "1714003300.000200", "text": "bot update" })) + .as_deref(), + Some("1714003300.000200") + ); + // Unparseable ts → dropped. + assert_eq!( + s.item_dedup_key(&json!({ "ts": "nope", "text": "hi" })), + None + ); + // Missing ts → dropped. + assert_eq!(s.item_dedup_key(&json!({ "text": "hi" })), None); + } + + #[test] + fn item_sort_ts_reads_raw_ts() { + let s = source(); + assert_eq!( + s.item_sort_ts(&json!({ "ts": "1714003200.000100" })) + .as_deref(), + Some("1714003200.000100") + ); + assert_eq!(s.item_sort_ts(&json!({ "text": "no ts" })), None); + } + + #[test] + fn advance_scope_cursor_merges_into_per_channel_map() { + let s = source(); + let mut state = SyncState::new("slack", "conn1"); + state.cursor = Some(r#"{"C1":"1714003200.000100"}"#.to_string()); + + s.advance_scope_cursor( + &mut state, + &SyncScope::nested("C2", "#two"), + "1714010000.000200", + ); + let map = sync::decode_cursors(state.cursor.as_deref()); + assert_eq!(map.get("C1").map(String::as_str), Some("1714003200.000100")); + assert_eq!(map.get("C2").map(String::as_str), Some("1714010000.000200")); + + // Re-advancing an existing channel overwrites just that entry. + s.advance_scope_cursor( + &mut state, + &SyncScope::nested("C1", "#one"), + "1714099999.000300", + ); + let map = sync::decode_cursors(state.cursor.as_deref()); + assert_eq!(map.get("C1").map(String::as_str), Some("1714099999.000300")); + assert_eq!(map.len(), 2); + } + + #[test] + fn source_advertises_slack_outlier_hooks() { + let s = source(); + assert!(s.per_scope_cursors()); + assert!(s.tolerate_scope_errors()); + assert!(s.server_side_depth()); + assert_eq!(s.toolkit(), "slack"); + assert_eq!(s.detail_noun(), "messages"); + assert_eq!(s.page_size(SyncReason::Periodic), HISTORY_PAGE_SIZE); + assert_eq!(s.max_pages(), MAX_HISTORY_PAGES_PER_CHANNEL); + } +} diff --git a/tests/memory_sync_slack_bus_raw_coverage_e2e.rs b/tests/memory_sync_slack_bus_raw_coverage_e2e.rs index c153ac2cc..8d4093f96 100644 --- a/tests/memory_sync_slack_bus_raw_coverage_e2e.rs +++ b/tests/memory_sync_slack_bus_raw_coverage_e2e.rs @@ -299,7 +299,10 @@ async fn slack_full_sync_search_backfill_and_bus_use_loopback_composio() { assert_eq!(outcome.toolkit, "slack"); assert_eq!(outcome.connection_id.as_deref(), Some("conn-slack-round19")); assert_eq!(outcome.items_ingested, 4); - assert_eq!(outcome.details["channels_processed"], 2); + // Slack now rides the generic orchestrator: two channels synced cleanly, + // none errored. (`channels_processed` → orchestrator's `scopes_synced`.) + assert_eq!(outcome.details["scopes_synced"], 2); + assert_eq!(outcome.details["scopes_errored"], 0); let search = run_backfill_via_search(&ctx, 2) .await